market_sync/app/tasks/task_moneyflow.py

"""同步任务：个股资金流（mairui）。

数据源：mairui `hsstock/history/transaction/{code}.{ex}/{licence}`
口径：主力 / 大 / 中 / 小单 净额（与 akshare stock_individual_fund_flow 一致）

akshare 已移除（项目级决策：永远不用），mairui 替代为唯一源。
"""
from __future__ import annotations

import os
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any

from app.core.db import ops as db_ops
from app.core.datasource.base import registry as ds_registry
from app.core.datasource.registry import is_source_ready
from app.core.datasource.utils import is_a_share_code, to_code6, to_hermes
from app.core.sync.base import SyncTask
from app.core.sync.registry import mark_sync_blocked
from app.core.utils.logging import get_logger

logger = get_logger("sync.moneyflow")


class SyncMoneyflow(SyncTask):
    dataset_id = "moneyflow"

    def _run(
        self,
        *,
        trigger_source: str = "manual",
        codes: list[str] | None = None,
        max_workers: int = 5,
        **kwargs,
    ) -> dict[str, Any]:
        mr = ds_registry.get("datasource_mairui")
        if mr is None:
            return {"status": "error", "message": "mairui 数据源未注册"}
        ok, reason = is_source_ready(mr.key)
        if not ok:
            mark_sync_blocked(self.dataset_id, message=f"mairui 未就绪: {reason}")
            return {"status": "blocked", "message": f"mairui 未就绪: {reason}"}

        if codes:
            stock_codes = [to_code6(c) for c in codes]
        else:
            stock_codes = [c for c in db_ops.iter_stock_codes(active_only=True) if is_a_share_code(c)]
        limit_raw = os.environ.get("MARKET_DATA_STOCK_LIMIT", "").strip()
        if limit_raw.isdigit() and int(limit_raw) > 0:
            stock_codes = stock_codes[: int(limit_raw)]

        if not stock_codes:
            return {"status": "error", "message": "无股票代码"}

        total = len(stock_codes)
        ok_cnt = fail_cnt = 0
        rows_total = 0
        t0 = time.time()
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures = {pool.submit(self._sync_one, mr, c6): c6 for c6 in stock_codes}
            for i, future in enumerate(as_completed(futures), 1):
                c6 = futures[future]
                try:
                    res = future.result()
                    if res["status"] == "ok":
                        ok_cnt += 1
                        rows_total += res["rows"]
                    else:
                        fail_cnt += 1
                        if res.get("error"):
                            logger.warning(f"[moneyflow {c6}] {res['error']}")
                except Exception as e:
                    fail_cnt += 1
                    logger.warning(f"[moneyflow {c6}] {e}")
                if i % 20 == 0 or i == total:
                    self._progress(
                        message=f"moneyflow 进度 {i}/{total} OK:{ok_cnt} FAIL:{fail_cnt}",
                        current=i, total=total, current_step=c6,
                    )
        elapsed = round(time.time() - t0, 1)
        msg = f"资金流 {ok_cnt}成 {fail_cnt}败 共{rows_total}行, {elapsed}s"
        return {
            "status": "ok" if fail_cnt == 0 else "warning",
            "message": msg,
            "ok": ok_cnt, "fail": fail_cnt,
            "rows": rows_total, "elapsed_sec": elapsed,
        }

    def _sync_one(self, mr, code6: str) -> dict:
        # 取最近 30 天（mairui 的 transaction 接口历史深度有限，30 天足够覆盖日常）
        from datetime import datetime, timedelta
        end = datetime.now()
        start = end - timedelta(days=30)
        try:
            df = mr.fetch_moneyflow(
                code6,
                start.strftime("%Y-%m-%d"),
                end.strftime("%Y-%m-%d"),
            )
            if df is None or df.empty:
                return {"status": "fail", "error": "no data"}
            rows = [
                {
                    "stock_code": to_hermes(code6),
                    "trade_date": str(r["trade_date"]),
                    "main_net_inflow": float(r.get("main_net_inflow") or 0),
                    "large_net_inflow": float(r.get("large_net_inflow") or 0),
                    "medium_net_inflow": float(r.get("medium_net_inflow") or 0),
                    "small_net_inflow": float(r.get("small_net_inflow") or 0),
                }
                for _, r in df.iterrows()
            ]
            db_ops.upsert_moneyflow(rows)
            return {"status": "ok", "rows": len(rows)}
        except Exception as e:
            return {"status": "fail", "error": str(e)}