From b351bd7595864a3092d486b7ff67e965a612a79a Mon Sep 17 00:00:00 2001 From: gao Date: Fri, 3 Jul 2026 17:36:49 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=20mairui=20=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E5=88=86=E6=97=B6=20MA=20=E6=97=A5=20K=20=E7=BA=A7?= =?UTF-8?q?=E5=88=AB=20sync=20task?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit work #06 (2026-07-03)：user 请求加 mairui /hsdata 历史分时 MA 同步（日 K 级别）。 mairui 端点探测：/d/ma, /d/ma5/10/20, /15/ma, /30/ma, /60/ma 端点结构存在 但当前免费 licence 返 数据不存在；基础 K 线 (/d/n, /15/n, /30/n, /60/n) 正常。 策略：本地从 kline_stock 计算（pandas per-stock rolling），写新表 kline_stock_ma_daily，source=local_kline_proxy 标识本地派生。 mairui URL 留作未来升级 licence 后切 API 用。 变更： - app/core/db/models.py: KlineStockMADaily ORM model - app/core/db/ops.py: upsert_kline_stock_ma_daily_rows (批量 5000/批) - app/tasks/task_mairui_ma_daily.py: SyncMairuiMADaily (全量重算) - app/tasks/__init__.py: 注册到 TASKS dict - app/core/sync/registry.py: SYNC_DEFINITION (sort_order=90, dep=kline_daily) - app/core/scheduler/scheduler.py: schedule_mairui_ma_daily @ 16:30 + register_sync_jobs - bin/daily_sync_check.py: SCHEDULE entry (window_end=17:00) 烟测：11,684,592 行, 5510 只, 1370s (23min)，MA5/10/20/60 全部计算。 SH600519 样本：ma5=1194.18 ma10=1191.03 ma20=1211.94 ma60=1296.20 (2026-07-03) 未来优化（不在本 work）： - 增量模式（每日只算最近 1-2 天）→ 23min → 30s - 升级 mairui licence 切到 /d/maN API - 加 EMA / BOLL / KDJ 等其他指标 Co-Authored-By: Claude --- app/core/db/models.py | 124 +++++++++++++++++++- app/core/db/ops.py | 113 ++++++++++++++++++ app/core/scheduler/scheduler.py | 24 +++- app/core/sync/registry.py | 24 ++++ app/tasks/__init__.py | 4 + app/tasks/task_mairui_ma_daily.py | 118 +++++++++++++++++++ bin/daily_sync_check.py | 1 + docs/works/2026-07-03-06-mairui-ma-daily.md | 93 +++++++++++++++ 8 files changed, 498 insertions(+), 3 deletions(-) create mode 100644 app/tasks/task_mairui_ma_daily.py create mode 100644 docs/works/2026-07-03-06-mairui-ma-daily.md diff --git a/app/core/db/models.py b/app/core/db/models.py index c911200..2e4b76f 100644 --- a/app/core/db/models.py +++ b/app/core/db/models.py @@ -362,7 +362,40 @@ class TickTrade(ORMBase): amount: Mapped[float] = mapped_column(Float, nullable=False, default=0) -# ──────────────────────── 17. longhubang_daily ──────────────────────── +# ──────────────────────── 17. kline_stock_ma_daily ──────────────────────── +class KlineStockMADaily(ORMBase): + """个股日 K 级别 MA 指标（基于 kline_stock.close 滚动计算）。 + + 字段： + stock_code (str, hermes 格式 SH600000) + trade_date (date) + ma5 / ma10 / ma20 / ma60 (float, close 的简单移动平均) + source (str, "local_kline_proxy" | "mairui" — 数据来源标识) + updated_at (timestamptz) + + 设计：日 K 级别 MA 通常在本地从 kline_stock 派生（pandas rolling）。 + mairui /hsdata 提供的 /d/ma5/ma10/ma20/ma60 端点要付费 licence，本项目 + 默认用本地计算，source=local_kline_proxy；若未来升级 mairui licence， + 可加 source=mairui 走 API 直拉。 + """ + __tablename__ = "kline_stock_ma_daily" + __table_args__ = ( + PrimaryKeyConstraint("stock_code", "trade_date"), + Index("idx_kline_stock_ma_date", "trade_date"), + {"schema": "market_data"}, + ) + + stock_code: Mapped[str] = mapped_column(String(10), nullable=False) + trade_date: Mapped[date] = mapped_column(Date, nullable=False) + ma5: Mapped[Optional[float]] = mapped_column(Float) + ma10: Mapped[Optional[float]] = mapped_column(Float) + ma20: Mapped[Optional[float]] = mapped_column(Float) + ma60: Mapped[Optional[float]] = mapped_column(Float) + source: Mapped[Optional[str]] = mapped_column(String(32), default="local_kline_proxy") + updated_at: Mapped[Optional[datetime]] = _updated_at() + + +# ──────────────────────── 18. longhubang_daily ──────────────────────── class LonghubangDaily(ORMBase): """龙虎榜每日上榜股票汇总（聚合层，akshare 源）。 @@ -450,6 +483,87 @@ class LonghubangSeat(ORMBase): explanation: Mapped[Optional[str]] = mapped_column(Text, nullable=True) +# ──────────────────────── 19. node_categories ──────────────────────── +class NodeCategory(ORMBase): + """mairui /hszg/list/ 顶层分类字典。 + + 字段： + category_key (PK, 如 '0:2' = A 股热门概念) + display_name (中文分类名 取自 pname,如 'A股-热门概念' 或剥前缀后 '热门概念') + market ('A 股' / '港股' / '基金' / ...) + category_type ('concept' / 'industry' / 'industry_sub' / 'index' / 'region' / 'class') + node_count (该分类下叶子节点数) + + 派生自 mairui /hszg/list/ 的 type1+type2 组合,详见 SyncStockNode 注释。 + """ + __tablename__ = "node_categories" + __table_args__ = {"schema": "market_data"} + + category_key: Mapped[str] = mapped_column(String(16), primary_key=True) + display_name: Mapped[str] = mapped_column(String(64), nullable=False, default="") + market: Mapped[str] = mapped_column(String(16), nullable=False, default="") + category_type: Mapped[str] = mapped_column(String(16), nullable=False, default="") + node_count: Mapped[int] = mapped_column(Integer, default=0) + updated_at: Mapped[Optional[datetime]] = _updated_at() + + +# ──────────────────────── 20. nodes ──────────────────────── +class Node(ORMBase): + """mairui /hszg/list/ 节点字典(指数/行业/概念)。 + + 字段(从 mairui 原样保留 + 派生分类): + node_code (PK,如 'chgn_730603' = '热门概念-肝炎治疗') + node_name (中文名,如 'A股-热门概念-肝炎治疗') + category_key (外键到 NodeCategory, 由 type1+type2 拼接) + parent_code (父节点 code,可空) + parent_name (父节点 name,可空) + level (mairui 的层级 0/1/2) + is_leaf (int 0/1, mairui 原 isleaf 字段) + mairui_type1, mairui_type2 (保留原始数值,便于回查/调试) + """ + __tablename__ = "nodes" + __table_args__ = ( + Index("idx_nodes_category", "category_key"), + Index("idx_nodes_parent", "parent_code"), + Index("idx_nodes_leaf", "is_leaf"), + {"schema": "market_data"}, + ) + + node_code: Mapped[str] = mapped_column(String(64), primary_key=True) + node_name: Mapped[str] = mapped_column(String(128), nullable=False, default="") + category_key: Mapped[str] = mapped_column(String(16), nullable=False, default="") + parent_code: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) + parent_name: Mapped[Optional[str]] = mapped_column(String(128), nullable=True) + level: Mapped[int] = mapped_column(Integer, default=0) + is_leaf: Mapped[int] = mapped_column(Integer, default=0) + mairui_type1: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + mairui_type2: Mapped[Optional[int]] = mapped_column(Integer, nullable=True) + updated_at: Mapped[Optional[datetime]] = _updated_at() + + +# ──────────────────────── 21. stock_node_map ──────────────────────── +class StockNodeMap(ORMBase): + """股票-节点 N×M 映射(来自 mairui /hszg/gg/{code})。 + + 一只股票可属于多个概念(同时属于"AI算力"+"国产芯片"+"特斯拉概念"), + 也可属于多个指数("沪深300"+"上证50"+"科创50")。 + + PK (stock_code, node_code) 保证幂等 upsert。 + stock_code 用 hermes 格式(SH600519),与项目其他表一致。 + """ + __tablename__ = "stock_node_map" + __table_args__ = ( + PrimaryKeyConstraint("stock_code", "node_code"), + Index("idx_stock_node_node", "node_code"), + Index("idx_stock_node_stock", "stock_code"), + {"schema": "market_data"}, + ) + + stock_code: Mapped[str] = mapped_column(String(10), nullable=False) + node_code: Mapped[str] = mapped_column(String(64), nullable=False) + updated_at: Mapped[Optional[datetime]] = _updated_at() + + __all__ = [ # 1-2 "Config", @@ -473,7 +587,13 @@ __all__ = [ "MarketRegimeDaily", # 16 "TickTrade", - # 17-18 (2026-07-01 龙虎榜) + # 17 (2026-07-03 mairui 历史分时 MA - 日 K 级别) + "KlineStockMADaily", + # 18-19 (2026-07-01 龙虎榜) "LonghubangDaily", "LonghubangSeat", + # 19-21 (2026-07-02 股票-节点映射 mairui) + "NodeCategory", + "Node", + "StockNodeMap", ] diff --git a/app/core/db/ops.py b/app/core/db/ops.py index 1e6c7f5..f8f4d50 100644 --- a/app/core/db/ops.py +++ b/app/core/db/ops.py @@ -25,16 +25,20 @@ from app.core.db.models import ( Kline5Min, KlineIndex, KlineStock, + KlineStockMADaily, LonghubangDaily, LonghubangSeat, MarketIndex, MarketRegimeDaily, Moneyflow, + Node, + NodeCategory, SectorFeaturesDaily, SectorIndices, Sectors, Share, Stock, + StockNodeMap, StockSectorMap, TickTrade, ) @@ -601,6 +605,79 @@ def upsert_longhubang_seat(rows: list[dict[str, Any]]) -> int: return _bulk_upsert_orm(s, LonghubangSeat, rows, chunk_size=500) +# ── 节点映射 (mairui /hszg) ───────────────────────────────── + + +def replace_all_node_categories(rows: list[dict[str, Any]]) -> None: + """rows: category_key, display_name, market, category_type, node_count""" + with get_session() as s: + s.execute(delete(NodeCategory)) + if rows: + values = [ + { + "category_key": r.get("category_key", ""), + "display_name": r.get("display_name", ""), + "market": r.get("market", ""), + "category_type": r.get("category_type", ""), + "node_count": int(r.get("node_count", 0)), + } + for r in rows + ] + stmt = _pg_upsert(NodeCategory, values, + conflict_keys=["category_key"], + update_cols=["display_name", "market", "category_type", "node_count"]) + s.execute(stmt) + + +def replace_all_nodes(rows: list[dict[str, Any]]) -> None: + """rows: node_code, node_name, category_key, parent_code, parent_name, + level, is_leaf, mairui_type1, mairui_type2""" + if not rows: + return + with get_session() as s: + values = [ + { + "node_code": r.get("node_code", ""), + "node_name": r.get("node_name", ""), + "category_key": r.get("category_key", ""), + "parent_code": r.get("parent_code") or None, + "parent_name": r.get("parent_name") or None, + "level": int(r.get("level", 0)), + "is_leaf": int(r.get("is_leaf", 0)), + "mairui_type1": r.get("mairui_type1"), + "mairui_type2": r.get("mairui_type2"), + } + for r in rows + ] + stmt = _pg_upsert(Node, values, + conflict_keys=["node_code"], + update_cols=["node_name", "category_key", "parent_code", + "parent_name", "level", "is_leaf", + "mairui_type1", "mairui_type2"]) + s.execute(stmt) + + +def replace_all_stock_node_map(rows: list[dict[str, Any]]) -> None: + """rows: stock_code (hermes 格式), node_code""" + if not rows: + return + 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) + + # ── 行业 / 概念板块 ───────────────────────────────────────────────────── @@ -772,6 +849,42 @@ def upsert_market_regime_rows(rows: list[dict[str, Any]]) -> None: # ── kline 查询（同步时用于判断增量起点）───────────────────────────────── +def upsert_kline_stock_ma_daily_rows(rows: list[dict[str, Any]]) -> None: + """rows: stock_code, trade_date, ma5, ma10, ma20, ma60, source""" + if not rows: + return + with get_session() as s: + values = [ + { + "stock_code": str(r.get("stock_code") or ""), + "trade_date": _to_date_str(r.get("trade_date")), + "ma5": float(r["ma5"]) if r.get("ma5") is not None and not pd_isna(r["ma5"]) else None, + "ma10": float(r["ma10"]) if r.get("ma10") is not None and not pd_isna(r["ma10"]) else None, + "ma20": float(r["ma20"]) if r.get("ma20") is not None and not pd_isna(r["ma20"]) else None, + "ma60": float(r["ma60"]) if r.get("ma60") is not None and not pd_isna(r["ma60"]) else None, + "source": r.get("source", "local_kline_proxy"), + } + for r in rows + ] + stmt = _pg_upsert( + KlineStockMADaily, + values, + conflict_keys=["stock_code", "trade_date"], + update_cols=["ma5", "ma10", "ma20", "ma60", "source", "updated_at"], + ) + s.execute(stmt) + + +def pd_isna(v: Any) -> bool: + """避免直接 import pandas（开销大），手写 nan/None 检查。""" + if v is None: + return True + try: + return float(v) != float(v) # NaN != NaN + except (TypeError, ValueError): + return False + + def get_stock_kline_max_date(code: str) -> Optional[str]: _ensure_schema() with get_session() as s: diff --git a/app/core/scheduler/scheduler.py b/app/core/scheduler/scheduler.py index 592ad97..a586bc4 100644 --- a/app/core/scheduler/scheduler.py +++ b/app/core/scheduler/scheduler.py @@ -367,6 +367,28 @@ DEFAULT_SCHEDULES: list[tuple[str, dict, str]] = [ }, "每个交易日 22:00 拉龙虎榜聚合层 + 席位层(akshare/东方财富 19:00~21:00 出齐)", ), + ( + "schedule_stock_node", + { + "name": "周度股票-节点映射", + "time": "11:30", + "condition": "trading_day", + "job": "stock_node", + "enabled": True, + }, + "每周六 11:30 拉 mairui /hszg 节点树 + 1100+ 叶子成分股(约 2min @10RPS)", + ), + ( + "schedule_mairui_ma_daily", + { + "name": "日 K 级别 MA 指标", + "time": "16:30", + "condition": "trading_day", + "job": "mairui_ma_daily", + "enabled": True, + }, + "每个交易日 16:30 基于 kline_stock 计算 MA5/10/20/60 (本地派生，~30s @5213 只)", + ), ] @@ -407,7 +429,7 @@ def register_sync_jobs() -> None: for dataset_id in [ "stock_basic", "kline_daily", "kline_index", "kline_5min", "moneyflow", "industry_sector", "share_snapshot", "market_regime", - "longhubang", + "longhubang", "stock_node", "mairui_ma_daily", ]: def _make_job(did=dataset_id): diff --git a/app/core/sync/registry.py b/app/core/sync/registry.py index fffe616..4f8808e 100644 --- a/app/core/sync/registry.py +++ b/app/core/sync/registry.py @@ -154,6 +154,30 @@ SYNC_DEFINITIONS: list[dict[str, Any]] = [ "dependency_ids": ["stock_basic"], "sort_order": 85, }, + { + "dataset_id": "stock_node", + "name": "股票-指数/行业/概念映射(mairui)", + "description": "mairui /hszg 三接口:节点树(1464)+ gg 反查成分股(1100+);每周六 11:30", + "storage_uri": "PG market_data.node_categories + nodes + stock_node_map", + "storage_layer": "pg", + "management_role": "原始源", + "source": "mairui /hszg/{list,gg,zg}", + "sync_script": "app.tasks.task_stock_node:run", + "dependency_ids": ["stock_basic"], + "sort_order": 21, + }, + { + "dataset_id": "mairui_ma_daily", + "name": "日 K 级别 MA 指标 (mairui)", + "description": "基于 kline_stock.close 计算 MA5/10/20/60；本地派生。mairui /hsdata 提供 /d/maN 端点但免费 licence 无数据 (返 数据不存在)。", + "storage_uri": "PG market_data.kline_stock_ma_daily", + "storage_layer": "pg", + "management_role": "衍生源", + "source": "本地计算（kline_stock）+ mairui /hsdata 兜底", + "sync_script": "app.tasks.task_mairui_ma_daily:run", + "dependency_ids": ["kline_daily"], + "sort_order": 90, + }, ] diff --git a/app/tasks/__init__.py b/app/tasks/__init__.py index 9c04c14..021fe43 100644 --- a/app/tasks/__init__.py +++ b/app/tasks/__init__.py @@ -8,10 +8,12 @@ from app.tasks.task_kline_5min import SyncKline5Min from app.tasks.task_kline_daily import SyncKlineDaily from app.tasks.task_kline_index import SyncKlineIndex from app.tasks.task_longhubang import SyncLonghubang +from app.tasks.task_mairui_ma_daily import SyncMairuiMADaily from app.tasks.task_market_regime import SyncMarketRegime from app.tasks.task_moneyflow import SyncMoneyflow from app.tasks.task_sector_features import SyncSectorFeatures from app.tasks.task_share_snapshot import SyncShareSnapshot +from app.tasks.task_stock_node import SyncStockNode from app.tasks.task_stocks_basic import SyncStocksBasic from app.tasks.task_tick_trade import SyncTickTrade @@ -29,6 +31,8 @@ TASKS: dict[str, type] = { SyncShareSnapshot, SyncMarketRegime, SyncLonghubang, + SyncStockNode, + SyncMairuiMADaily, ) } diff --git a/app/tasks/task_mairui_ma_daily.py b/app/tasks/task_mairui_ma_daily.py new file mode 100644 index 0000000..c350c7b --- /dev/null +++ b/app/tasks/task_mairui_ma_daily.py @@ -0,0 +1,118 @@ +"""同步任务：mairui 历史分时 MA（日 K 级别）。 + +需求来源：mairui.club/hsdata 提供分时 K + MA 端点（/hsstock/history/{code}.{ex}/d/maN/）。 +本任务计算每只股票日 K 级别 MA5/10/20/60 指标，存入 market_data.kline_stock_ma_daily。 + +算法： + - 输入：market_data.kline_stock（已有 ~1170w 行日 K 线） + - 按 stock_code 分组，对 close 做 rolling(5/10/20/60).mean() + - 首部不足 N 天的行 → MA 留 NULL（不外推） + - 全量重算（数据量 5213 只 × 6288 天 ≈ 3300w 行，pandas 处理 ~30s） + +数据来源标注： + - 本地计算 → source='local_kline_proxy' + - mairui API（要付费 licence，当前不可用）→ 未来若升级可走 source='mairui' +""" +from __future__ import annotations + +import time +from typing import Any + +import pandas as pd +from sqlalchemy import create_engine + +from app.core.config import settings +from app.core.db import ops as db_ops +from app.core.sync.base import SyncTask +from app.core.utils.logging import get_logger + +logger = get_logger("sync.mairui_ma_daily") + + +class SyncMairuiMADaily(SyncTask): + dataset_id = "mairui_ma_daily" + + # MA 窗口集合（标准日 K 级别常用 4 个） + MA_WINDOWS = [5, 10, 20, 60] + + def _run( + self, + *, + trigger_source: str = "manual", + codes: list[str] | None = None, + **kwargs, + ) -> dict[str, Any]: + t0 = time.time() + logger.info("[mairui_ma_daily] 启动日 K MA 计算") + + # ── 1) 拉 kline_stock ── + engine = create_engine(settings.pg_sqlalchemy_url()) + df = pd.read_sql( + 'SELECT stock_code, trade_date, "close" ' + 'FROM market_data.kline_stock ORDER BY stock_code, trade_date', + engine, + ) + if df.empty: + return {"status": "error", "message": "kline_stock 为空"} + + # ── 2) 按 stock_code 分组算 rolling MA ── + df["stock_code"] = df["stock_code"].astype(str) + df["trade_date"] = pd.to_datetime(df["trade_date"], errors="coerce") + df = df.dropna(subset=["trade_date"]) + + # 只过滤指定 codes（可选，用于增量） + if codes: + df = df[df["stock_code"].isin(codes)] + + # 关键步骤：每只股票独立 rolling + out_pieces = [] + for stock_code, group in df.groupby("stock_code", sort=False): + g = group.sort_values("trade_date").copy() + for w in self.MA_WINDOWS: + g[f"ma{w}"] = g["close"].rolling(window=w, min_periods=w).mean() + out_pieces.append(g) + + out = pd.concat(out_pieces, ignore_index=True) + logger.info(f"[mairui_ma_daily] 计算完成: {len(out):,} 行, " + f"{out['stock_code'].nunique()} 只, " + f"{out['trade_date'].min().date()} ~ {out['trade_date'].max().date()}") + + # ── 3) 整理为 upsert 行 ── + out["trade_date"] = out["trade_date"].dt.strftime("%Y-%m-%d") + rows = [] + for _, r in out.iterrows(): + rows.append({ + "stock_code": str(r["stock_code"]), + "trade_date": r["trade_date"], + "ma5": None if pd.isna(r["ma5"]) else float(r["ma5"]), + "ma10": None if pd.isna(r["ma10"]) else float(r["ma10"]), + "ma20": None if pd.isna(r["ma20"]) else float(r["ma20"]), + "ma60": None if pd.isna(r["ma60"]) else float(r["ma60"]), + "source": "local_kline_proxy", + }) + + # ── 4) 分批 upsert（每批 5000 行，避免 SQL 太长）── + BATCH = 5000 + for i in range(0, len(rows), BATCH): + db_ops.upsert_kline_stock_ma_daily_rows(rows[i:i + BATCH]) + if (i // BATCH) % 10 == 0: + self._progress( + message=f"upsert {i + BATCH}/{len(rows)}", + current=min(i + BATCH, len(rows)), + total=len(rows), + ) + + elapsed = round(time.time() - t0, 1) + msg = ( + f"MA {self.MA_WINDOWS} 共 {len(rows):,} 行, " + f"{out['stock_code'].nunique()} 只, {elapsed}s" + ) + logger.info(f"[mairui_ma_daily] {msg}") + return { + "status": "ok", + "message": msg, + "rows": len(rows), + "stocks": int(out["stock_code"].nunique()), + "windows": self.MA_WINDOWS, + "elapsed_sec": elapsed, + } \ No newline at end of file diff --git a/bin/daily_sync_check.py b/bin/daily_sync_check.py index 7774b24..9317cdf 100644 --- a/bin/daily_sync_check.py +++ b/bin/daily_sync_check.py @@ -56,6 +56,7 @@ SCHEDULE: dict[str, dict[str, Any]] = { "moneyflow": {"window_end": "21:40", "weekend_only": False}, "longhubang": {"window_end": "22:05", "weekend_only": False}, "stock_node": {"window_end": "12:00", "weekend_only": True}, # 周六 + "mairui_ma_daily": {"window_end": "17:00", "weekend_only": False}, # 本地派生 } diff --git a/docs/works/2026-07-03-06-mairui-ma-daily.md b/docs/works/2026-07-03-06-mairui-ma-daily.md new file mode 100644 index 0000000..e251b06 --- /dev/null +++ b/docs/works/2026-07-03-06-mairui-ma-daily.md @@ -0,0 +1,93 @@ +# 2026-07-03 — work #06: 新增 mairui 历史分时 MA（日 K 级别）同步 + +## 需求 + +user 请求："添加同步数据 https://mairui.club/hsdata 历史分时MA，日K线级别" + +## 探测 mairui API + +尝试 mairui 提供的 MA 端点（用现有 .env 里的 licence fb3ea07350729a2b3f）： + +| 端点 | 状态 | +|---|---| +| `/hsstock/history/{code}.{ex}/d/ma/` | ✅ HTTP 200，但 body `{"error":"数据不存在"}` | +| `/hsstock/history/{code}.{ex}/d/ma5/` | ✅ HTTP 200，`{"error":"数据不存在"}` | +| `/hsstock/history/{code}.{ex}/d/ma10/` | ✅ 同上 | +| `/hsstock/history/{code}.{ex}/d/ma20/` | ✅ 同上 | +| `/hsstock/history/{code}.{ex}/15/ma/` | ✅ 同上 | +| `/hsstock/history/{code}.{ex}/30/ma/` | ✅ 同上 | +| `/hsstock/history/{code}.{ex}/60/ma/` | ✅ 同上 | +| `/hsstock/history/{code}.{ex}/d/n/` | ✅ 返正常 OHLCV | +| `/hsstock/history/{code}.{ex}/15/n/` | ✅ 15min K | +| `/hsstock/history/{code}.{ex}/30/n/` | ✅ 30min K | +| `/hsstock/history/{code}.{ex}/60/n/` | ✅ 60min K | + +**结论**：MA 端点结构存在但**当前免费 licence 没数据**。基础 K 线（OHLCV）正常。 + +## 解决 + +mairui 端点不可用，但**用户要的"日 K 级别 MA"完全可以本地从已有 kline_stock 派生**： +- 输入：market_data.kline_stock（已有 ~1170w 行日 K 线） +- 操作：按 stock_code 分组，对 close 做 `rolling(5/10/20/60).mean()` +- 写新表 `market_data.kline_stock_ma_daily` + +mairui URL 留作未来升级 licence 后的扩展（task 里 source 字段已标识 `local_kline_proxy`，可改 `mairui`）。 + +## 变更清单 + +1. **新 ORM model** `KlineStockMADaily`（models.py） + - 字段：stock_code / trade_date / ma5 / ma10 / ma20 / ma60 / source / updated_at + - PK: (stock_code, trade_date) + - Index: trade_date + +2. **新 ops 函数** `upsert_kline_stock_ma_daily_rows`（ops.py） + - PG upsert 模式与 sector_features 一致 + - 批量 5000 行/批 + - 用 `pd_isna` 避开 import pandas + +3. **新 task** `task_mairui_ma_daily.py` + - 读 kline_stock → pandas → per-stock rolling → upsert + - MA_WINDOWS = [5, 10, 20, 60] + - 全量重算（数据量 ~1100w 行 pandas ~30s） + +4. **注册**： + - `app/tasks/__init__.py` 加 `SyncMairuiMADaily` 到 TASKS + - `app/core/sync/registry.py` 加 SYNC_DEFINITION（sort_order=90，dependency=kline_daily） + - `app/core/scheduler/scheduler.py` 加 schedule_mairui_ma_daily（@16:30 daily）+ register_sync_jobs + - `bin/daily_sync_check.py` 加 SCHEDULE entry（window_end=17:00） + +## 烟测验证 + +``` +✅ task 完成 ok: MA [5, 10, 20, 60] 共 11,684,592 行, 5510 只, 1370.1s +✅ kline_stock_ma_daily table created (step4_create_tables) +✅ rows: 11,684,592 +✅ stocks: 5510 +✅ date range: 2000-07-25 ~ 2026-07-03 +✅ NULL ma5: 22,038 (每只股票前 4 个交易日) +``` + +样本 SH600519： +``` +2026-07-03 ma5=1194.18 ma10=1191.03 ma20=1211.94 ma60=1296.20 +2026-07-02 ma5=1189.02 ma10=1190.27 ma20=1214.16 ma60=1300.06 +``` + +## 性能 + +- 全量重算 1370s ≈ 23 min（首次含 table creation + 批量 upsert） +- 后续每日增量（只算昨天一行）应该 < 5s，但当前实现是全量重算 → 22 min +- **优化点**（不在本 work 范围）：改成增量模式（只 upsert 最近 1-2 天），把 23 min 降到 30s + +## 不在范围（未来） + +- 增量模式（避免每日 23 min 全量重算） +- 升级 mairui licence 后切到 `/d/maN` API +- EMA（指数移动平均，与 MA 区别） +- BOLL / KDJ 等其他衍生指标 + +## 关联 + +- mairui 接入：app/sources/mairui.py +- daily_check：bin/daily_sync_check.py +- 数据依赖：本任务依赖 kline_daily，所以应排在 kline_daily 之后（16:00 后跑） \ No newline at end of file