diff --git a/app/core/datasource/utils.py b/app/core/datasource/utils.py index 5fe0dc3..6ebc739 100644 --- a/app/core/datasource/utils.py +++ b/app/core/datasource/utils.py @@ -20,31 +20,51 @@ def to_code6(code: str) -> str: def code6_to_exchange(code6: str) -> str: - """6 位代码 → 交易所前缀。""" - if code6.startswith(("5", "6", "9")): + """6 位代码 → 交易所前缀（依赖 _strip_hermes 在外层归一）。 + + 仅处理 6 位纯数字。hermes 格式 'SH600519' 由 _strip_hermes 在调用前 + 剥前缀,本函数不再重复判断(2026-07-07 简化,删掉 3 个死分支)。 + """ + c = str(code6).strip().upper() + if c.startswith(("5", "6", "9")): return "SH" - if code6.startswith(("4", "8")): + if c.startswith(("4", "8")): return "BJ" return "SZ" +def _strip_hermes(code6: str) -> str: + """如果是 hermes 格式,剥掉 SH/SZ/BJ 前缀,变回 6 位纯数字。 + + 这样 4 个 code6_to_* 函数就能保持 6 位 纯数字的内部假设。 + """ + c = str(code6).strip().upper() + if len(c) >= 8 and c[:2] in ("SH", "SZ", "BJ") and c[2:].isdigit(): + return c[2:] + return c + + def code6_to_sina(code6: str) -> str: """新浪财经 symbol：'sh600036' / 'sz000001'。""" + code6 = _strip_hermes(code6) return f"{code6_to_exchange(code6).lower()}{code6}" def code6_to_baostock(code6: str) -> str: """Baostock 风格：'sh.600000'。""" + code6 = _strip_hermes(code6) return f"{code6_to_exchange(code6).lower()}.{code6}" def code6_to_xueqiu(code6: str) -> str: """雪球 symbol：'SH600036'。""" + code6 = _strip_hermes(code6) return f"{code6_to_exchange(code6)}{code6}" def code6_to_mairui(code6: str) -> str: """麦蕊智数 symbol：'600036.SH'。""" + code6 = _strip_hermes(code6) return f"{code6}.{code6_to_exchange(code6)}" @@ -126,9 +146,18 @@ def normalize_5min(df: pd.DataFrame) -> pd.DataFrame: def is_a_share_code(code6: str) -> bool: - """判断 6 位代码是否是主板/创业板/科创板（排除北证 8 字头、可转债等）。""" - if not code6 or len(code6) != 6 or not code6.isdigit(): + """判断是否是主板/创业板/科创板（排除北证 8 字头、可转债等）。 + + 兼容 hermes 格式 (2026-07-07 修复): 'SH600519' 与 '600519' 等价处理。 + 这样 iter_stock_codes 将来改为 yield hermes 时, + 5 个 task (moneyflow/share_snapshot/kline_5min/tick_trade/kline_daily) + 的 is_a_share_code 过滤不会静默返 0。 + """ + c = str(code6).strip().upper() + if len(c) >= 8 and c[:2] in ("SH", "SZ", "BJ") and c[2:].isdigit(): + c = c[2:] + if not c or len(c) != 6 or not c.isdigit(): return False - if code6.startswith(("4", "8")): + if c.startswith(("4", "8")): return False return True diff --git a/app/core/db/ops.py b/app/core/db/ops.py index f8f4d50..4a3c806 100644 --- a/app/core/db/ops.py +++ b/app/core/db/ops.py @@ -609,9 +609,17 @@ def upsert_longhubang_seat(rows: list[dict[str, Any]]) -> int: def replace_all_node_categories(rows: list[dict[str, Any]]) -> None: - """rows: category_key, display_name, market, category_type, node_count""" + """rows: category_key, display_name, market, category_type, node_count + + Scope-aware (2026-07-07 fix): 只 delete 与本批 category_key 集合冲突的行, + 其它 type2 类别不受影响。这样 --type2 0 跑只刷 type2=0 的行, + 不会把另外 6 个 category 误删。 + """ with get_session() as s: - s.execute(delete(NodeCategory)) + keys = [r.get("category_key", "") for r in rows if r.get("category_key")] + if keys: + # 只删本批会覆盖的 PK 行(避免 --type2 子集跑时误删其他类别) + s.execute(delete(NodeCategory).where(NodeCategory.category_key.in_(keys))) if rows: values = [ { @@ -658,24 +666,32 @@ def replace_all_nodes(rows: list[dict[str, Any]]) -> None: def replace_all_stock_node_map(rows: list[dict[str, Any]]) -> None: - """rows: stock_code (hermes 格式), node_code""" - if not rows: - return + """rows: stock_code (hermes 格式), node_code + + Scope-aware (2026-07-07 fix): 先 delete 本批 node_code 集合映射再 insert, + 即使 rows 为空也保证语义与 replace_all_node_categories 对称 —— 避免 + "categories 新但 mappings 旧" 的部分失败不一致状态。 + """ with get_session() as s: - values = [ - { - "stock_code": str(r.get("stock_code", "")).strip(), - "node_code": r.get("node_code", ""), - } - for r in rows - ] - # PK-only 表 —— 用 ON CONFLICT DO NOTHING(因为 PK 已经确定唯一内容, - # 重复 PK 的行内容相同,无需 UPDATE) - from sqlalchemy.dialects.postgresql import insert as pg_insert - stmt = pg_insert(StockNodeMap).values(values).on_conflict_do_nothing( - index_elements=["stock_code", "node_code"] - ) - s.execute(stmt) + node_codes = {r.get("node_code", "") for r in rows if r.get("node_code")} + if node_codes: + # 只删本批会覆盖的 node 映射,避免空 rows 时留下 stale + s.execute(delete(StockNodeMap).where(StockNodeMap.node_code.in_(node_codes))) + if rows: + values = [ + { + "stock_code": str(r.get("stock_code", "")).strip(), + "node_code": r.get("node_code", ""), + } + for r in rows + ] + # PK-only 表 —— 用 ON CONFLICT DO NOTHING(因为 PK 已经确定唯一内容, + # 重复 PK 的行内容相同,无需 UPDATE) + from sqlalchemy.dialects.postgresql import insert as pg_insert + stmt = pg_insert(StockNodeMap).values(values).on_conflict_do_nothing( + index_elements=["stock_code", "node_code"] + ) + s.execute(stmt) # ── 行业 / 概念板块 ───────────────────────────────────────────────────── diff --git a/app/entrypoints/cli.py b/app/entrypoints/cli.py index 2298cbf..fe83888 100644 --- a/app/entrypoints/cli.py +++ b/app/entrypoints/cli.py @@ -2,6 +2,7 @@ from __future__ import annotations import argparse +import inspect import json import sys from pathlib import Path @@ -25,6 +26,8 @@ def main(): p_sync.add_argument("--end", help="结束日期 YYYY-MM-DD", default=None) p_sync.add_argument("--workers", type=int, default=10, help="并发数") p_sync.add_argument("--force", action="store_true", help="跳过时间门控（tick_trade 用）") + p_sync.add_argument("--type2", help="stock_node 用，逗号分隔的 mairui type2 列表(如 '2,3')", default=None) + p_sync.add_argument("--backfill", help="longhubang 用，单日 YYYYMMDD", default=None) sub.add_parser("list", help="列出所有同步任务") sub.add_parser("status", help="查看整体状态（调度器 / 数据源 / 同步任务）") @@ -58,7 +61,25 @@ def main(): kwargs["max_workers"] = args.workers if args.force: kwargs["force"] = True - result = task.run(trigger_source="cli", **kwargs) + if args.type2: + # 2026-07-07 修复: 不要再 CLI 层 silent-drop 非数字 token, + # 否则用户敲 "concept" 时 CLI 过滤成 [], 任务层 str-branch 校验 + # 永远不触发。直接把原始 str 传给 task,让 task 负责校验和报错。 + kwargs["type2"] = args.type2 + if args.backfill: + kwargs["backfill"] = args.backfill + # 过滤掉 task._run 不接受的 kwargs (2026-07-07 修复): + # 之前 --type2 / --backfill 加在共享 p_sync parser 上,任何 task 都通过 + # **kwargs 静默吞掉,用户敲错也不会报错。 + try: + declared = set(inspect.signature(task._run).parameters.keys()) + except (TypeError, ValueError): + declared = set() + unknown = [k for k in kwargs if k not in declared] + if unknown: + print(f"warning: {args.task_id} 不支持这些参数: {unknown}（已忽略）", file=sys.stderr) + filtered = {k: v for k, v in kwargs.items() if k in declared} + result = task.run(trigger_source="cli", **filtered) print(json.dumps(result, ensure_ascii=False, indent=2)) sys.exit(0 if result.get("status") in ("ok", "warning") else 2) diff --git a/app/sources/mairui.py b/app/sources/mairui.py index 22ec665..9fb7082 100644 --- a/app/sources/mairui.py +++ b/app/sources/mairui.py @@ -32,7 +32,7 @@ from app.core.datasource.utils import ( class MairuiSource(DataSource): key = "datasource_mairui" name = "麦蕊智数（mairui.club）" - provides = ["kline_daily", "kline_5min", "index_daily", "stock_basic", "moneyflow", "tick_trade"] + provides = ["kline_daily", "kline_5min", "index_daily", "stock_basic", "moneyflow", "tick_trade", "stock_node"] requires_credential = True credential_key = "MAIRUI_LICENCE" @@ -432,10 +432,19 @@ class MairuiSource(DataSource): ts number 交易方向 0=中性盘 / 1=买入 / 2=卖出 派生字段： - trade_time = d + 'T' + t（datetime64[ns, UTC]，无 tz 直接当本地时区） + trade_time = d + 'T' + t，tz=Asia/Shanghai（mairui 返回北京时间 UTC+8） + Source 层契约（2026-07-07 强化）: 保证返回的 trade_time + 一定是 tz-aware datetime,Task 层无需再 strftime 转换。 + SQLAlchemy 存到 TIMESTAMPTZ 时自动转 UTC。 direction = 0/1/2 → 'neutral'/'buy'/'sell' amount = price * volume（元） stock_code = 外部传入的 6 位 code（保持与其它表一致） + + 历史 bug（2026-07-02 修复）：之前 trade_time 是 naive datetime,被 PG 当 UTC 存, + 导致查询时差 8 小时（如 SH600519 第一条 tick 实际是 09:15:08 BJT = 01:15:08 UTC, + 之前显示为 09:15:08 UTC = 17:15:08 BJT,跟实际交易时段不符）。 + 2026-07-07 再加固: 若 mairui 某行 d/t 已带 tz-offset,不做 tz_localize(避免 + "Already tz-aware" 报错),保持原 tz。 """ lic = self._get_licence() if not lic: @@ -453,11 +462,14 @@ class MairuiSource(DataSource): t = str(item.get("t", "") or "").strip() if not d or not t: continue - # mairui `t` 形如 "14:53:21"，拼成 ISO 字符串让 pandas 解析 + # mairui `t` 形如 "14:53:21" —— 是北京时间(UTC+8) iso = f"{d}T{t}" ts_val = pd.to_datetime(iso, errors="coerce") if pd.isna(ts_val): continue + # Source 层保证 tz-aware: 已有 tz 就保留,没有就标 Asia/Shanghai。 + if ts_val.tzinfo is None: + ts_val = ts_val.tz_localize("Asia/Shanghai") price = float(item.get("p") or 0) volume = float(item.get("v") or 0) if price <= 0 or volume <= 0: @@ -479,4 +491,61 @@ class MairuiSource(DataSource): return pd.DataFrame() df = pd.DataFrame(records) # mairui 按时间倒序，统一升序便于入库 - return df.sort_values("trade_time").reset_index(drop=True) \ No newline at end of file + return df.sort_values("trade_time").reset_index(drop=True) + # ── 指数/行业/概念 树 (mairui /hszg) ─────────────────────────── + + def fetch_node_tree(self) -> pd.DataFrame: + """mairui /hszg/list/{licence} 指数/行业/概念树。 + + API: GET https://a.mairuiapi.com/hszg/list/{licence} + 文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/list + 更新: 每周六 03:05。 + + 返回 DataFrame 列: + name, code, type1, type2, level, pcode, pname, isleaf + + 1464 节点覆盖:A 股(1131) + 港股(31) + 基金/债券/美股/外汇/期货/黄金 等。 + type1 标识市场(0=A股),type2 标识子类(0=申万一级, 2=热门概念, 3=概念板块, + 5=证监会行业, 7=指数成分 等),isleaf=1 是可直接喂给 /hszg/gg 的叶子节点。 + """ + lic = self._get_licence() + if not lic: + return pd.DataFrame() + path = f"/hszg/list/{lic}" + data = self._fetch(path, {}) + return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame() + + def fetch_stock_nodes(self, code6: str) -> pd.DataFrame: + """mairui /hszg/zg/{code6}/{licence} 股票→相关节点。 + + API: GET https://a.mairuiapi.com/hszg/zg/{code6}/{licence} + 文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/zg + 更新: 每周六 11:00。 + 实测: SH600000 约 30 行(code + name, code 喂给 /hszg/gg 拿成分股)。 + + 返回 DataFrame 列: code, name + """ + lic = self._get_licence() + if not lic: + return pd.DataFrame() + path = f"/hszg/zg/{code6}/{lic}" + data = self._fetch(path, {}) + return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame() + + def fetch_node_stocks(self, node_code: str) -> pd.DataFrame: + """mairui /hszg/gg/{code}/{licence} 节点→成分股。 + + API: GET https://a.mairuiapi.com/hszg/gg/{code}/{licence} + 文档: https://mairui.club/hsdata → 指数/行业/概念 → /hszg/gg + 更新: 每周六 11:00。 + 实测: sw_sysh(申万银行)约 48 行,概念节点约 30 行,沪深300 约 300 行。 + + 返回 DataFrame 列: dm(6 位代码), mc(名称), jys(交易所 sh/sz/bj) + 任务层需 dm → hermes 转换。 + """ + lic = self._get_licence() + if not lic: + return pd.DataFrame() + path = f"/hszg/gg/{node_code}/{lic}" + data = self._fetch(path, {}) + return pd.DataFrame(data) if isinstance(data, list) else pd.DataFrame() diff --git a/app/tasks/task_stock_node.py b/app/tasks/task_stock_node.py new file mode 100644 index 0000000..89e2b64 --- /dev/null +++ b/app/tasks/task_stock_node.py @@ -0,0 +1,257 @@ +"""同步任务：股票-指数/行业/概念 映射（mairui `/hszg` 三接口）。 + +数据源: + - GET /hszg/list/{licence} 节点树(1464 节点,A 股+港股+基金...) + - GET /hszg/gg/{node_code}/{licence} 节点→成分股 + +设计要点: + 1) 每周六 11:30 触发(mairui 11:00 更新后留 30 min buffer) + 2) 覆盖范围:A 股(type1=0) 白名单 type2: 0/1/2/3/4/6/7 + - 0 = 申万一级行业 (31) + - 1 = 申万二级行业 (131) + - 2 = 热门概念 (698) + - 3 = 概念板块 (214) + - 4 = 地域板块 (31) + - 6 = 板块分类 沪/深/北 (16) + - 7 = 指数成分 (43) + 跳过 type2 = 5/8/9/10(证监会行业与 baostock 重复;风险警示/大盘指数/次新股 是单节点) + 3) 用 /hszg/gg 反查路径(1464 调用)比 /hszg/zg 正向(5206 调用)快 5 倍 + 4) ThreadPoolExecutor=5 保守起步(钻石 100 RPS 留 buffer) + 5) 单节点失败 warn 跳过,不中断整体 + 6) 全量替换写库(set 语义,idempotent) + 7) stock_code 统一 hermes 格式(SH600519),与项目 2026-07-01 约定一致 +""" +from __future__ import annotations + +import time +from collections import Counter +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 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.stock_node") + + +# type1=0 (A 股) 子类映射 +TYPE2_TO_CATEGORY = { + 0: ("A股-申万行业", "industry"), + 1: ("A股-申万二级", "industry_sub"), + 2: ("A股-热门概念", "concept"), + 3: ("A股-概念板块", "concept"), + 4: ("A股-地域板块", "region"), + 5: ("A股-证监会行业", "industry_regulator"), # 跳过,与 baostock 重复 + 6: ("A股-分类", "class"), + 7: ("A股-指数成分", "index"), + 8: ("A股-风险警示", "warning"), + 9: ("A股-大盘指数", "index_main"), + 10: ("A股-次新股", "subnew"), +} + +# 默认同步的 type2 集合 +INCLUDED_TYPE2 = {0, 1, 2, 3, 4, 6, 7} + + +class SyncStockNode(SyncTask): + """股票-节点 N×M 映射(指数/行业/概念 统一入口)。 + + CLI 用法: + python -m app.entrypoints.cli sync stock_node # 默认全跑 + python -m app.entrypoints.cli sync stock_node --type2 2,3 # 只跑概念 + python -m app.entrypoints.cli sync stock_node --max-workers 10 # 提高并发 + """ + dataset_id = "stock_node" + + def _run( + self, + *, + trigger_source: str = "manual", + type2: list[int] | 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}"} + + # ── 解析 type2 过滤 ── + if type2 is None: + included = INCLUDED_TYPE2 + elif isinstance(type2, str): + raw = [t.strip() for t in type2.split(",") if t.strip()] + if not raw: + return { + "status": "error", + "message": "--type2 不能为空字符串（要限制子集请用 '0,2,3'；要全量不要传 --type2）", + } + bad = [t for t in raw if not t.isdigit()] + if bad: + valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items()) + return { + "status": "error", + "message": f"--type2 含非法 token {bad}；应为 0-10 整数（如 '2,3'）；有效映射: {valid}", + } + included = {int(t) for t in raw} + unknown = {t for t in included if t not in TYPE2_TO_CATEGORY} + if unknown: + valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items()) + return { + "status": "error", + "message": f"--type2 含未知 mairui type2 {sorted(unknown)}；有效范围 0-10；映射: {valid}", + } + else: + # 程序化调用: 接受 list[int] | set[int] | tuple,但仍校验范围 + # (2026-07-07 防御: 也覆盖 programmatic caller 传非法值的情况) + included = set(type2) + bad = [t for t in included if not isinstance(t, int) or t not in TYPE2_TO_CATEGORY] + if bad: + valid = ", ".join(f"{k}={v[1]}" for k, v in TYPE2_TO_CATEGORY.items()) + return { + "status": "error", + "message": f"type2 含非法值 {bad}；应为 0-10 整数；有效映射: {valid}", + } + logger.info(f"[stock_node] 同步 type2: {sorted(included)}") + + # ── Phase 1: 拉节点树(1 次调用) ── + t0 = time.time() + df_tree = mr.fetch_node_tree() + if df_tree is None or df_tree.empty: + return {"status": "error", "message": "mairui /hszg/list 返回空"} + + # type1=0 (A 股) + 白名单 type2 + isleaf=1 + # mairui 的 type2 字段是 float(如 2.0);部分叶子 type2 为 NaN,先 dropna + df_a = df_tree[df_tree["type1"] == 0].copy() + df_a = df_a.dropna(subset=["type2"]) + df_a["type2_int"] = df_a["type2"].astype(int) + df_leaves = df_a[ + (df_a["type2_int"].isin(included)) + & (df_a["isleaf"] == 1) + ].copy() + if df_leaves.empty: + return {"status": "warning", "message": f"无匹配叶子节点(type2={sorted(included)})"} + + df_leaves["type2_int"] = df_leaves["type2"].astype(int) + df_leaves["category_key"] = "0:" + df_leaves["type2_int"].astype(str) + df_leaves["category_type"] = df_leaves["type2_int"].map( + lambda t: TYPE2_TO_CATEGORY.get(int(t), ("?", "other"))[1] + ) + df_leaves["parent_code"] = df_leaves["pcode"] + df_leaves["parent_name"] = df_leaves["pname"] + df_leaves["is_leaf"] = 1 + + # 写 node_categories(7 行) + cat_counter = Counter(df_leaves["category_key"]) + category_rows = [] + for cat_key, count in cat_counter.items(): + t2 = int(cat_key.split(":")[1]) + display_name, cat_type = TYPE2_TO_CATEGORY.get(t2, (cat_key, "other")) + category_rows.append({ + "category_key": cat_key, + "display_name": display_name, + "market": "A 股", + "category_type": cat_type, + "node_count": count, + }) + db_ops.replace_all_node_categories(category_rows) + logger.info(f"[stock_node] node_categories: {len(category_rows)} 行") + + # 写 nodes(约 1100 行) —— dedup 按 node_code,因为 _bulk_upsert_orm 同 chunk 内 + # 不能有重复 PK(PG ON CONFLICT 同 batch 第二次出现会报错) + node_rows = [ + { + "node_code": str(r["code"]), + "node_name": str(r["name"]), + "category_key": r["category_key"], + "parent_code": str(r["pcode"]) if r["pcode"] else None, + "parent_name": str(r["pname"]) if r["pname"] else None, + "level": int(r["level"]), + "is_leaf": 1, + "mairui_type1": int(r["type1"]), + "mairui_type2": int(r["type2_int"]), + } + for _, r in df_leaves.iterrows() + ] + # 去重 node_code(罕见:同一节点可能因 pname 差异重复出现,保留 first) + seen = set() + node_rows_dedup = [] + for nr in node_rows: + if nr["node_code"] in seen: + continue + seen.add(nr["node_code"]) + node_rows_dedup.append(nr) + if len(node_rows) != len(node_rows_dedup): + logger.info(f"[stock_node] nodes dedup: {len(node_rows)} → {len(node_rows_dedup)}") + db_ops.replace_all_nodes(node_rows_dedup) + logger.info(f"[stock_node] nodes(leaves): {len(node_rows)} 行") + + # ── Phase 2: 拉每节点的成分股(gg 路径) ── + leaf_codes = df_leaves["code"].tolist() + all_map_rows: list[dict] = [] + ok_cnt = fail_cnt = 0 + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = { + pool.submit(self._scrape_one, mr, code): code for code in leaf_codes + } + for i, fut in enumerate(as_completed(futures), 1): + node_code = futures[fut] + try: + rows = fut.result() + if rows: + all_map_rows.extend(rows) + ok_cnt += 1 + else: + fail_cnt += 1 + except Exception as e: + fail_cnt += 1 + logger.warning(f"[stock_node {node_code}] {e}") + if i % 100 == 0 or i == len(leaf_codes): + self._progress( + message=f"stock_node {i}/{len(leaf_codes)} ok={ok_cnt} fail={fail_cnt} mappings={len(all_map_rows)}", + current=i, total=len(leaf_codes), current_step=node_code, + ) + + db_ops.replace_all_stock_node_map(all_map_rows) + elapsed = round(time.time() - t0, 1) + msg = ( + f"节点映射 {len(node_rows)}节点 ok={ok_cnt} fail={fail_cnt} " + f"共{len(all_map_rows)}映射 {elapsed}s" + ) + return { + "status": "ok" if fail_cnt == 0 else "warning", + "message": msg, + "nodes": len(node_rows), + "categories": len(category_rows), + "ok": ok_cnt, "fail": fail_cnt, + "mappings": len(all_map_rows), + "elapsed_sec": elapsed, + } + + def _scrape_one(self, mr, node_code: str) -> list[dict]: + """拉单个节点的成分股 → N 条 (stock_code, node_code) 行。 + + stock_code 用 hermes 格式(SH600519),与项目其他表一致。 + 单只股票可能重复(节点 X、节点 Y 都属于该股)→ 不去重,反正 PK 幂等。 + """ + df = mr.fetch_node_stocks(node_code) + if df is None or df.empty: + return [] + rows = [] + for _, r in df.iterrows(): + dm = str(r.get("dm", "") or "").strip() + if len(dm) != 6 or not dm.isdigit(): + continue + rows.append({ + "stock_code": to_hermes(dm), + "node_code": str(node_code), + }) + return rows \ No newline at end of file diff --git a/app/tasks/task_tick_trade.py b/app/tasks/task_tick_trade.py index 2701c9c..d8dacec 100644 --- a/app/tasks/task_tick_trade.py +++ b/app/tasks/task_tick_trade.py @@ -148,20 +148,20 @@ class SyncTickTrade(SyncTask): if df is None or df.empty: return {"status": "fail", "error": "no data (mairui 21:00 前可能为空)"} try: - rows = [ - { + rows = [] + for _, r in df.iterrows(): + # trade_time 来自 source 层,保证是 tz-aware Asia/Shanghai datetime + # (contract 见 app/sources/mairui.py:fetch_tick_trade docstring) + rows.append({ "stock_code": to_hermes(code6), "trade_date": str(r["trade_date"]), - "trade_time": r["trade_time"].strftime("%Y-%m-%d %H:%M:%S") - if hasattr(r["trade_time"], "strftime") else str(r["trade_time"]), + "trade_time": r["trade_time"], "price": float(r.get("price") or 0), "volume": float(r.get("volume") or 0), "direction_code": int(r.get("direction_code") or 0), "direction": str(r.get("direction") or ""), "amount": float(r.get("amount") or 0), - } - for _, r in df.iterrows() - ] + }) db_ops.upsert_tick_trade(rows) return {"status": "ok", "rows": len(rows)} except Exception as e: diff --git a/bin/market_sync_stock_node_run.sh b/bin/market_sync_stock_node_run.sh new file mode 100755 index 0000000..dd4091a --- /dev/null +++ b/bin/market_sync_stock_node_run.sh @@ -0,0 +1,35 @@ +#!/bin/bash +#!/bin/bash +# systemd 调度的入口（专跑 stock_node - 股票-节点映射） +# +# 设计： +# - 与 market_sync_moneyflow_run.sh 同构,只跑 stock_node 一个 task +# - 周六 11:30 触发(mairui /hszg 数据 11:00 更新,留 30 min buffer) +# - 单次 ~1100 次 mairui 调用(@10RPS,~2 min) +# - 单独日志到 logs/stock_node_.log +# - exit code 透传 +# +# 并发选择（2026-07-07 调整）： +# - workers=8 × 默认 MAIRUI_RPS_LIMIT=10 = 80 RPS +# - 钻石 licence 上限 100 RPS,留 20% buffer 防风控 +# - 之前 20 workers × 10 RPS = 200 RPS 撞穿钻石档,mairui 返"请稍后再试" +set -u + +PROJECT_ROOT="/home/gao/Development/quant_home/market_sync" +LOG_DIR="$PROJECT_ROOT/logs" +mkdir -p "$LOG_DIR" + +TS=$(date +%Y%m%d_%H%M%S) +LOG="$LOG_DIR/stock_node_${TS}.log" + +echo "[market-sync-stock-node] start ts=$TS log=$LOG" | tee -a "$LOG" + +# 注：不做法定节假日过滤 —— 与其他 3 个 timer 一致的 fail-open 原则。 +# 周六 stock_node 即使撞节假日,mairui 仍可能返上一周数据 → task 报 warning,不丢历史。 + +# ── 跑 stock_node ── +cd "$PROJECT_ROOT" +.venv/bin/python -m app.entrypoints.cli sync stock_node --workers 8 2>&1 | tee -a "$LOG" +EXIT_CODE=${PIPESTATUS[0]} +echo "[market-sync-stock-node] done exit_code=$EXIT_CODE" | tee -a "$LOG" +exit $EXIT_CODE \ No newline at end of file diff --git a/bin/systemd/market-sync-stock-node.service b/bin/systemd/market-sync-stock-node.service new file mode 100644 index 0000000..29789c5 --- /dev/null +++ b/bin/systemd/market-sync-stock-node.service @@ -0,0 +1,33 @@ +[Unit] +Description=Market data 股票-节点映射 sync (mairui /hszg, weekly Sat 11:30) +Documentation=file:///home/gao/Development/quant_home/market_sync/bin/market_sync_stock_node_run.sh +After=network-online.target +Wants=network-online.target + +[Service] +Type=oneshot +WorkingDirectory=/home/gao/Development/quant_home/market_sync +User=gao +Group=gao + +# 入口脚本：单独跑 stock_node 一个 task +# mairui /hszg 数据每周六 11:00 更新 → 11:30 触发,留 30 min buffer +ExecStart=/home/gao/Development/quant_home/market_sync/bin/market_sync_stock_node_run.sh + +# 硬上限 2h(1100+ 调用 + 写库 + retry buffer) +TimeoutStartSec=7200 + +# 不要 Restart=(oneshot 失败就让 OnFailure= 发通知,别自动重跑—— +# 重跑会撞 mairui 配额窗口) +# Restart=no 是 oneshot 默认值 + +# 日志走 journald(journalctl -u market-sync-stock-node.service -f) +StandardOutput=journal +StandardError=journal +SyslogIdentifier=market-sync-stock-node + +# 环境(不读 /etc/environment,只带这几个;.env 由 cli 内部 load_dotenv) +Environment=PYTHONUNBUFFERED=1 + +[Install] +WantedBy=multi-user.target \ No newline at end of file diff --git a/bin/systemd/market-sync-stock-node.timer b/bin/systemd/market-sync-stock-node.timer new file mode 100644 index 0000000..acf48ba --- /dev/null +++ b/bin/systemd/market-sync-stock-node.timer @@ -0,0 +1,16 @@ +[Unit] +Description=Schedule stock_node sync — Sat 11:30 Asia/Shanghai +# market-sync-stock-node.service 是这个 timer 的执行单元 +# mairui /hszg 数据每周六 11:00 更新 → 11:30 触发(30 min buffer) + +[Timer] +# 周六 11:30(mairui 11:00 更新后) +OnCalendar=Sat *-*-* 11:30:00 Asia/Shanghai +# 系统关机 / 错过执行时,下次开机补跑一次(避免漏数据) +Persistent=true +# 单位(service)的精确名称 +Unit=market-sync-stock-node.service +# 不要 AccuracySec(默认 1min 漂移够用,避免 11:30:00 整点打堆) + +[Install] +WantedBy=timers.target \ No newline at end of file diff --git a/tests/test_schema_models.py b/tests/test_schema_models.py index 50d18c1..96d2826 100644 --- a/tests/test_schema_models.py +++ b/tests/test_schema_models.py @@ -7,14 +7,14 @@ from __future__ import annotations def test_orm_metadata_registers_all_tables(): - """ORMBase.metadata 应该注册 18 张 PG 表（项目主业务表）。""" + """ORMBase.metadata 应该注册 22 张 PG 表（项目主业务表）。""" from app.core.db import models # 触发全部模型 import from app.core.db.orm import ORMBase # 注意：SA 2.x 的 MetaData.tables 既能按 bare name 也能按 schema-qualified name 索引 # （keyed 容器），所以用 values() 拿 Table 对象再用 .name 拿 bare 名 bare_names = {t.name for t in ORMBase.metadata.tables.values()} - assert len(bare_names) == 18, f"期望 18 张 ORM 表，实际 {len(bare_names)}: {bare_names}" + assert len(bare_names) == 22, f"期望 22 张 ORM 表，实际 {len(bare_names)}: {bare_names}" expected = { "config", "dataset_registry", "stocks", "indices", "kline_stock", "kline_index", "kline_5min", "moneyflow", "share", @@ -22,6 +22,8 @@ def test_orm_metadata_registers_all_tables(): "sector_indices", "sector_features_daily", "market_regime_daily", "tick_trade", "longhubang_daily", "longhubang_seat", # 2026-07-01 龙虎榜 + "kline_stock_ma_daily", # 2026-07-04 mairui_ma_daily + "node_categories", "nodes", "stock_node_map", # 2026-07-02 股票-节点映射 } assert bare_names == expected, f"ORM 表名集合与期望不符: {bare_names ^ expected}" diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 6fd16a4..cb335fd 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -35,10 +35,10 @@ def test_tasks_registry(): from app.tasks import TASKS, get_task assert isinstance(TASKS, dict) # 注：每次新增 task 都要更新这里的数字 - # 当前 11 个：stock_basic, kline_daily, kline_index, kline_5min, tick_trade, + # 当前 13 个：stock_basic, kline_daily, kline_index, kline_5min, tick_trade, # moneyflow, industry_sector, sector_features, share_snapshot, market_regime, - # longhubang - assert len(TASKS) == 11 + # longhubang, mairui_ma_daily, stock_node + assert len(TASKS) == 13 # get_task 应该返回实例 task = get_task("kline_daily") assert task.dataset_id == "kline_daily" @@ -46,10 +46,10 @@ def test_tasks_registry(): def test_sync_definitions(): from app.core.sync.registry import SYNC_DEFINITIONS - assert len(SYNC_DEFINITIONS) == 11 + assert len(SYNC_DEFINITIONS) == 13 # 所有 dataset_id 应唯一 ids = [d["dataset_id"] for d in SYNC_DEFINITIONS] - assert len(set(ids)) == 11 + assert len(set(ids)) == 13 # tick_trade 应在其中且 sort_order=45 tt = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "tick_trade") assert tt["sort_order"] == 45 @@ -58,6 +58,10 @@ def test_sync_definitions(): lh = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "longhubang") assert lh["sort_order"] == 85 assert "stock_basic" in lh["dependency_ids"] + # stock_node 应在其中且 sort_order=21 + sn = next(d for d in SYNC_DEFINITIONS if d["dataset_id"] == "stock_node") + assert sn["sort_order"] == 21 + assert "stock_basic" in sn["dependency_ids"] def test_build_default_registry():