diff --git a/scripts/data_platform/ak_events_wrapper.ps1 b/scripts/data_platform/ak_events_wrapper.ps1 index a8c217e..16b7a5d 100644 --- a/scripts/data_platform/ak_events_wrapper.ps1 +++ b/scripts/data_platform/ak_events_wrapper.ps1 @@ -1,6 +1,7 @@ # ak_events_wrapper.ps1 — sanguo-ak-events schtask wrapper (daily 19:30 akshare 事件类 per-date 当日) -# dragon_tiger/block_trade/margin_sse/restricted/zt_pool 三件套 --start today --end today; -# per-date 每类1 unit 快; zt_pool 系 (2026-09-02 双机实测 push2ex 集群稳定) 供情绪/打板因子 +# dragon_tiger/block_trade/margin_sse/restricted/zt_pool 三件套/新浪资金流×2/同花顺行业概念 --start today --end today; +# per-date 每类1 unit 快; zt_pool 系 (2026-09-02 双机实测 push2ex 集群稳定) 供情绪/打板因子; +# 新浪资金流+同花顺榜 (2026-09-02 用户批 A 档: 东财 push2 墙死后免费无墙替代, 秒级零限流) $env:http_proxy = '' $env:https_proxy = '' $env:all_proxy = '' @@ -10,5 +11,5 @@ $ts = Get-Date -Format 'yyyyMMdd_HHmmss' $logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } $log = Join-Path $logDir "ak_events_$ts.txt" -C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types dragon_tiger,block_trade,margin_sse,restricted,zt_pool,zt_pool_zbgc,zt_pool_dtgc --start $today --end $today *>> $log +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\akshare_static_download.py --types dragon_tiger,block_trade,margin_sse,restricted,zt_pool,zt_pool_zbgc,zt_pool_dtgc,fund_flow_industry,fund_flow_concept,ths_industry,ths_concept --start $today --end $today *>> $log exit $LASTEXITCODE diff --git a/scripts/data_platform/akshare_static_download.py b/scripts/data_platform/akshare_static_download.py index 45cbf36..7e6c1ee 100644 --- a/scripts/data_platform/akshare_static_download.py +++ b/scripts/data_platform/akshare_static_download.py @@ -143,7 +143,7 @@ PER_STOCK_TYPES = ( "financial_abstract", # stock_financial_abstract(symbol="600519") # top_holders 单独 (per-stock × per-period) ) -# 模式 B: per-date 类型 (7 类, margin_szse 跳过) +# 模式 B: per-date 类型 (11 类, margin_szse 跳过) PER_DATE_TYPES = ( "dragon_tiger", # stock_lhb_detail_em(start_date, end_date) "block_trade", # stock_dzjy_mrmx(symbol="A股", start_date, end_date) @@ -156,6 +156,14 @@ PER_DATE_TYPES = ( "zt_pool", # stock_zt_pool_em(date) 涨停池 "zt_pool_zbgc", # stock_zt_pool_zbgc_em(date) 炸板池 "zt_pool_dtgc", # stock_zt_pool_dtgc_em(date) 跌停池 + # 行业/概念截面快照 (2026-09-02 用户批 A 档接入; 东财 push2 clist 墙死后的 + # 免费/无墙替代源, 双机实测秒级零限流)。⚠️ 快照型端点无 date 参数、无历史 + # 回补, date 仅命名文件 (19:30 采集=收盘终态); 节假日拉到上一交易日快照 + # (无自日期列可判), 接受重复, 消费方按交易日历过滤。 + "fund_flow_industry", # 新浪 stock_fund_flow_industry('即时') 行业资金流 90 行 (净额/领涨股) + "fund_flow_concept", # 新浪 stock_fund_flow_concept('即时') 概念资金流 ~387 行 + "ths_industry", # 同花顺 stock_board_industry_summary_ths() 行业一览 90 行×12 列 + "ths_concept", # 同花顺 stock_board_concept_name_ths() 概念名单 ~375 行 ) # 模式 C: per-period 类型 (2 类) PER_PERIOD_TYPES = ( @@ -642,6 +650,38 @@ def fetch_zt_pool_dtgc(date: str) -> Optional[pd.DataFrame]: return df +def fetch_fund_flow_industry(date: str) -> Optional[pd.DataFrame]: + """stock_fund_flow_industry(symbol='即时') — 新浪行业资金流快照 (90 行)。""" + df, _status = call_ak_with_retry( + ak.stock_fund_flow_industry, f"fund_flow_industry/{date}", symbol="即时", + ) + return df + + +def fetch_fund_flow_concept(date: str) -> Optional[pd.DataFrame]: + """stock_fund_flow_concept(symbol='即时') — 新浪概念资金流快照 (~387 行)。""" + df, _status = call_ak_with_retry( + ak.stock_fund_flow_concept, f"fund_flow_concept/{date}", symbol="即时", + ) + return df + + +def fetch_ths_industry(date: str) -> Optional[pd.DataFrame]: + """stock_board_industry_summary_ths() — 同花顺行业一览快照 (90 行×12 列)。""" + df, _status = call_ak_with_retry( + ak.stock_board_industry_summary_ths, f"ths_industry/{date}", + ) + return df + + +def fetch_ths_concept(date: str) -> Optional[pd.DataFrame]: + """stock_board_concept_name_ths() — 同花顺概念名单快照 (~375 行)。""" + df, _status = call_ak_with_retry( + ak.stock_board_concept_name_ths, f"ths_concept/{date}", + ) + return df + + # ======================== per-period fetch 函数 (2 类) ======================== def fetch_forecast(period: str) -> Optional[pd.DataFrame]: @@ -1085,6 +1125,10 @@ def run_type_dispatch( "zt_pool": fetch_zt_pool, "zt_pool_zbgc": fetch_zt_pool_zbgc, "zt_pool_dtgc": fetch_zt_pool_dtgc, + "fund_flow_industry": fetch_fund_flow_industry, + "fund_flow_concept": fetch_fund_flow_concept, + "ths_industry": fetch_ths_industry, + "ths_concept": fetch_ths_concept, } units = build_per_date_units(t, fetch_map[t], args) return run_one_type(t, units, args) diff --git a/scripts/data_platform/static_vintage_check.py b/scripts/data_platform/static_vintage_check.py index 5b971e4..a018243 100644 --- a/scripts/data_platform/static_vintage_check.py +++ b/scripts/data_platform/static_vintage_check.py @@ -32,7 +32,14 @@ EMPTY_PARQUET_MIN_BYTES = 1024 # 与 akshare_static_download 同口径 # 涨停池三件套 (2026-09-02 入列 ak-events): per-date 家族每个工作日(含节假日, # 空数据也写空文件)应有 1 个 parquet, 文件缺 = 当晚 schtask 漏跑 = 洞。 # 端点只支持最近 ~30 交易日 → 洞滑出窗口 = 永久缺 (不可逆), 09-02 与策略侧共识防线。 -PANEL_TYPES = ("zt_pool", "zt_pool_zbgc", "zt_pool_dtgc") +# 快照族 (新浪资金流/同花顺榜, 09-02 用户批 A 档接入): 端点无历史回补能力, +# 任何洞即时即永久 → 告警语义不同 (查当晚日志, 无回补指引)。 +PANEL_TYPES = ( + "zt_pool", "zt_pool_zbgc", "zt_pool_dtgc", # 可回补族 + "fund_flow_industry", "fund_flow_concept", # 快照族 (永久) + "ths_industry", "ths_concept", +) +BACKFILLABLE_PANEL_TYPES = frozenset({"zt_pool", "zt_pool_zbgc", "zt_pool_dtgc"}) PANEL_BACKFILL_CALENDAR_DAYS = 42 # ≈30 交易日回补窗的日历日近似 @@ -126,15 +133,19 @@ def panel_gap_check(root: Path) -> dict: continue entry = {"n_files": len(dates), "first_date": None, "holes_backfillable": [], "holes_permanent": []} + backfillable_type = t in BACKFILLABLE_PANEL_TYPES if dates: first = min(dates) entry["first_date"] = first.isoformat() cur = first while cur < today: if cur.weekday() < 5 and cur not in dates: - key = ("holes_backfillable" - if (today - cur).days <= PANEL_BACKFILL_CALENDAR_DAYS - else "holes_permanent") + # 快照族无回补能力, 任何洞即时即永久; 可回补族按 42 日历日分档 + recent = (today - cur).days <= PANEL_BACKFILL_CALENDAR_DAYS + if backfillable_type and recent: + key = "holes_backfillable" + else: + key = "holes_permanent" entry[key].append(cur.isoformat()) cur += datetime.timedelta(days=1) out[t] = entry @@ -180,7 +191,8 @@ def check_all(root: Path) -> Tuple[dict, list]: alerts.append( "估值表断更: 最新 mtime %s 距今 %.1fh (>%.0fh, ak-eod 未跑成?)" % (v["newest_mtime"], v["age_hours"], VALUATION_STALE_HOURS)) - # 涨停池三件套缺日 (09-02 入列): 可回补洞告警, 永久缺只入 json + # 涨停池三件套缺日 (09-02 入列): 可回补洞告警, 永久缺只入 json; + # 快照族 (新浪/同花顺) 近窗洞=昨晚漏跑, 洞即永久 → 告警无回补指引 status["panel"] = panel_gap_check(root) for t, e in status["panel"].items(): if e["holes_backfillable"]: @@ -188,6 +200,15 @@ def check_all(root: Path) -> Tuple[dict, list]: "[%s] 缺日(回补窗内): %s (当晚 schtask 漏跑; 30 交易日内手动 " "--start 缺日 --end 缺日 可补, 滑出窗=永久缺)" % ( t, ",".join(e["holes_backfillable"]))) + elif t not in BACKFILLABLE_PANEL_TYPES and e["holes_permanent"]: + fresh_holes = [ + h for h in e["holes_permanent"] + if (datetime.date.today() + - datetime.date.fromisoformat(h)).days <= 3] + if fresh_holes: + alerts.append( + "[%s] 缺日(快照族无回补, 洞即永久): %s (查当晚 migration_logs " + "定位漏跑原因)" % (t, ",".join(fresh_holes))) return status, alerts diff --git a/tests/data_platform/test_akshare_static_download.py b/tests/data_platform/test_akshare_static_download.py index 992fd8d..648965c 100644 --- a/tests/data_platform/test_akshare_static_download.py +++ b/tests/data_platform/test_akshare_static_download.py @@ -133,3 +133,28 @@ class TestZtPoolFamily: assert status == "empty" assert (tmp_path / "zt_pool_dtgc" / "20260901_zt_pool_dtgc.parquet").exists() assert (tmp_path / "zt_pool_dtgc" / ".20260901_zt_pool_dtgc.akshare").exists() + + +class TestSinaThsFamily: + """新浪资金流 + 同花顺行业/概念榜 (09-02 用户批 A 档, 东财墙的免费替代)。""" + + def test_registered_as_per_date_types(self): + for t in ("fund_flow_industry", "fund_flow_concept", + "ths_industry", "ths_concept"): + assert t in mod.PER_DATE_TYPES + assert t in mod.ALL_TYPES + + def test_fetchers_none_on_retry_exhausted(self, monkeypatch): + monkeypatch.setattr( + mod, "call_ak_with_retry", lambda *a, **k: (None, "failed")) + for fetch in (mod.fetch_fund_flow_industry, mod.fetch_fund_flow_concept, + mod.fetch_ths_industry, mod.fetch_ths_concept): + assert fetch("20260902") is None + + def test_fetchers_df_on_ok(self, monkeypatch): + df = pd.DataFrame({"净额": [1.0]}) + monkeypatch.setattr(mod, "call_ak_with_retry", + lambda *a, **k: (df, "ok")) + for fetch in (mod.fetch_fund_flow_industry, mod.fetch_fund_flow_concept, + mod.fetch_ths_industry, mod.fetch_ths_concept): + assert fetch("20260902") is df diff --git a/tests/data_platform/test_static_vintage_check.py b/tests/data_platform/test_static_vintage_check.py index e56f475..9cec1db 100644 --- a/tests/data_platform/test_static_vintage_check.py +++ b/tests/data_platform/test_static_vintage_check.py @@ -182,6 +182,24 @@ class TestPanelGapCheck: assert not any("缺日" in a for a in alerts) assert status["panel"]["zt_pool"]["n_files"] == 0 + def test_snapshot_family_hole_alerts_as_permanent(self, tmp_path): + """快照族 (无回补能力) 近窗洞 → 告警说『洞即永久/查日志』而非回补指引。""" + _mk_ok_statements(tmp_path) + hole = _recent_weekdays(2)[-1] + dates = _recent_weekdays(6, drop_dates=(hole,)) + d = tmp_path / "fund_flow_industry" + d.mkdir(parents=True) + for dt in dates: + (d / f"{dt:%Y%m%d}_fund_flow_industry.parquet").write_bytes(b"x" * 10) + _, alerts = svc.check_all(tmp_path) + assert any("fund_flow_industry" in a and "洞即永久" in a + and hole.isoformat() in a for a in alerts) + # 快照族近窗洞归 holes_permanent (无回补档) + status_panel = svc.panel_gap_check(tmp_path) + assert status_panel["fund_flow_industry"]["holes_backfillable"] == [] + assert hole.isoformat() in ( + status_panel["fund_flow_industry"]["holes_permanent"]) + # ======================== main / JSON 契约 ========================