diff --git a/scripts/data_platform/bs_daily_main.py b/scripts/data_platform/bs_daily_main.py new file mode 100644 index 0000000..a05b143 --- /dev/null +++ b/scripts/data_platform/bs_daily_main.py @@ -0,0 +1,71 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""bs_daily_main.py — sanguo-bs-daily (VPS schtask 18:05) 单入口: 串行 bs_eod → bs_fundamentals. + +2026-08-20 数据任务重组(用户拍板「bs 每天」合并): 根治当日撞点 —— bs_eod 主循环后 +新增指数段后单日跑 5~5.5h(ETA 23:00+), 与旧 sanguo-bs-fund 23:05 重叠 = 同 IP 双 +baostock 连接 = 封禁红线(6-24h)。合并进同一 schtask 后两腿先后子进程, 连接天然互斥: +leg1 进程退出(finally bs.logout + 进程消亡)连接才彻底释放, leg2 才 login。 + +编排原则(两脚本幂等/断点/配额逻辑原样复用零改动): +- 只做顺序 + 退出码透传 + 时间线日志([BS-DAILY] 腿起止行, 供验收"eod logout 之后 + fund 才 login"的直接证据) +- leg2 无条件跑: leg1 无论成败, 进程退出即连接释放; fund 是独立作业(自带登录重试/ + 日预算), leg1 致命不该连坐饿死当日财务回灌 +- 退出码语义沿用两脚本: 0=完成/让路, 1=致命, 2=登录失败, 3=日预算/防线 graceful stop + (3 是"按预算收工", 不算失败不上报); 任务级退出码 = 第一个非 OK 腿的码 + +配额(单 IP 48000/天铁律): wrapper 传 BS_DAILY_LIMIT=15000 收紧 eod 病态重试上限 +(常态 ~11.1k 不受影响) + fund 自有 DAILY_CAP=30000 → 最坏 45.1k < 48k。 +""" +import argparse +import logging +import subprocess +import sys +import time +from pathlib import Path + +HERE = Path(__file__).resolve().parent +LEGS = ( + ("bs_eod", HERE / "bs_eod.py"), + ("bs_fund", HERE / "bs_fundamentals.py"), +) +OK_CODES = {0, 3} # 3=按预算 graceful stop, 属正常收工 + +logging.basicConfig(level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + handlers=[logging.StreamHandler(sys.stdout)]) +log = logging.getLogger(__name__) + + +def run_leg(name, script): + t0 = time.time() + log.info("[BS-DAILY] leg=%s start script=%s", name, script.name) + rc = subprocess.run( + [sys.executable, "-X", "utf8", str(script)]).returncode + log.info("[BS-DAILY] leg=%s exit=%d 耗时%.0fs", name, rc, time.time() - t0) + return rc + + +def main(argv=None): + ap = argparse.ArgumentParser() + ap.add_argument("--smoke", action="store_true", + help="只验 schtask→wrapper→python 链路可达, 两腿不跑(零配额)") + args = ap.parse_args(argv) + if args.smoke: + log.info("[BS-DAILY] smoke ok 编排入口可达") + return 0 + + rc1 = run_leg(*LEGS[0]) + if rc1 not in OK_CODES: + log.warning("[BS-DAILY] leg1 非常规退出 %d —— 连接已随进程释放, " + "fund 独立作业照跑", rc1) + rc2 = run_leg(*LEGS[1]) + + if rc1 not in OK_CODES: + return rc1 + return 0 if rc2 in OK_CODES else rc2 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/data_platform/bs_daily_wrapper.ps1 b/scripts/data_platform/bs_daily_wrapper.ps1 new file mode 100644 index 0000000..dbf9e1b --- /dev/null +++ b/scripts/data_platform/bs_daily_wrapper.ps1 @@ -0,0 +1,17 @@ +# bs_daily_wrapper.ps1 — sanguo-bs-daily schtask wrapper (18:05 baostock 日任务单入口) +# 2026-08-20 重组: 串行 bs_eod(个股 EOD+15m+指数段) → bs_fundamentals(季频财务+业绩报告) +# —— 同一 schtask 内先后子进程, 同 IP 永不双 baostock 连接(根治 08-20 撞点: eod 加 +# 指数段后 ETA 23:00+ 与旧 23:05 bs-fund 重叠)。 +# BS_DAILY_LIMIT=15000: eod 病态重试硬顶(常态 ~11.1k 不受影响), 与 fund 30000 合计 +# ≤45.1k < 48k/IP/天(配额铁律)。 +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +$env:BS_DAILY_LIMIT = '15000' +Set-Location C:\sanguo_vnpy_v2 +$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 "bs_daily_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\bs_daily_main.py *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/bs_fundamentals.py b/scripts/data_platform/bs_fundamentals.py index 655acf1..c1d8654 100644 --- a/scripts/data_platform/bs_fundamentals.py +++ b/scripts/data_platform/bs_fundamentals.py @@ -277,6 +277,9 @@ def _bs_eod_running(): """sanguo-bs-eod schtask 是否仍在跑(同 IP 双 baostock 连接红线)。 探测失败按未在跑处理(fail-open)—— 23:05 距 bs_eod 极端尾 22:41 有 20min 余量。 + 2026-08-20 重组后: 本脚本改由 sanguo-bs-daily(bs_daily_main 串行编排)调用, + 互斥已结构性保证; 旧任务名 sanguo-bs-eod 已删, 探测必失败→fail-open 继续 + = 正确行为, 勿"修复"成探新任务名(会探到自己的父任务在跑而永久让路)。 """ try: out = subprocess.run( diff --git a/tests/data_platform/test_bs_daily_main.py b/tests/data_platform/test_bs_daily_main.py new file mode 100644 index 0000000..2688d9a --- /dev/null +++ b/tests/data_platform/test_bs_daily_main.py @@ -0,0 +1,87 @@ +# -*- coding: utf-8 -*- +"""TDD for bs_daily_main.py — sanguo-bs-daily 单入口串行编排 (2026-08-20 重组). + +核心契约: +1. 顺序铁律: bs_eod 腿必先于 bs_fund 腿(subprocess.run 阻塞=天然串行, 同 IP 永不 + 双 baostock 连接 —— 08-20 撞点(eod 加指数段 ETA 23:00+ 撞 fund 23:05)的根治) +2. leg2 无条件跑: leg1 致命(1)/登录失败(2)不连坐 —— 连接已随进程释放, fund 独立作业 +3. 退出码聚合: {0,3}=OK(3=按预算 graceful stop 不算失败), 任务码=第一个非 OK 腿 +4. --smoke: 零配额链路自检, 不起任何子进程 +""" +import sys +from unittest.mock import MagicMock, patch + +import pytest + +if "baostock" not in sys.modules: # bs_fundamentals import 链上需要 + sys.modules["baostock"] = MagicMock() + +from scripts.data_platform import bs_daily_main as bd # noqa: E402 + + +def _mock_run(rc1, rc2): + """subprocess.run 按调用次序返回 returncode=rc1/rc2 的假结果, 记录调用脚本名。""" + calls = [] + + def fake_run(cmd, **kw): + calls.append(cmd[-1]) + return MagicMock(returncode=rc1 if len(calls) == 1 else rc2) + + return calls, fake_run + + +# ---------- smoke ---------- + +def test_smoke_exits_0_without_subprocess(): + with patch.object(bd.subprocess, "run", + side_effect=AssertionError("smoke 不应起子进程")): + assert bd.main(["--smoke"]) == 0 + + +# ---------- 顺序铁律 ---------- + +def test_runs_eod_then_fund_in_order(): + calls, fake = _mock_run(0, 0) + with patch.object(bd.subprocess, "run", side_effect=fake): + assert bd.main([]) == 0 + assert [c.rsplit("\\", 1)[-1].rsplit("/", 1)[-1] for c in calls] == \ + ["bs_eod.py", "bs_fundamentals.py"] + + +def test_leg_scripts_resolve_to_real_files(): + for _name, script in bd.LEGS: + assert script.exists(), script + + +# ---------- leg2 无条件跑(leg1 异常不连坐) ---------- + +@pytest.mark.parametrize("rc1", [1, 2]) +def test_leg2_runs_even_when_leg1_fatal(rc1): + calls, fake = _mock_run(rc1, 0) + with patch.object(bd.subprocess, "run", side_effect=fake): + bd.main([]) + assert len(calls) == 2 # fund 腿照跑 + + +# ---------- 退出码聚合 ---------- + +@pytest.mark.parametrize("rc1,rc2,expected", [ + (0, 0, 0), + (3, 0, 0), # eod 按预算 graceful stop = 正常 + (0, 3, 0), # fund 按日预算收工 = 正常 + (3, 3, 0), + (1, 0, 1), # leg1 致命 → 任务码如实上报(不被 leg2 的 0 掩盖) + (2, 0, 2), + (0, 1, 1), + (0, 2, 2), + (1, 2, 1), # 双异常取第一个(先修 leg1) +]) +def test_exit_aggregation(rc1, rc2, expected): + _calls, fake = _mock_run(rc1, rc2) + with patch.object(bd.subprocess, "run", side_effect=fake): + assert bd.main([]) == expected + + +def test_ok_codes_contract(): + """{0,3}=OK 契约钉死: 两脚本退出码语义变更时这里必须同步评审。""" + assert bd.OK_CODES == {0, 3}