feat(data): sanguo-bs-daily单入口串行编排(bs_eod→bs_fundamentals)——2026-08-20数据任务重组阶段1:根治当日撞点(bs_eod加指数段后ETA 23:00+与旧23:05 bs-fund重叠=同IP双baostock连接封禁红线),合并进同一schtask两腿先后子进程,leg1进程退出(logout+消亡)连接才释放leg2才login;bs_daily_main只做顺序+退出码透传({0,3}=OK,3=按预算graceful不算失败)+[BS-DAILY]腿起止时间线日志(验收'eod logout后fund才login'直接证据);leg2无条件跑(leg1致命不连坐,连接已释放fund独立作业);wrapper传BS_DAILY_LIMIT=15000收紧eod病态重试硬顶(常态~11.1k不受影响,与fund30000合计≤45.1k<48k/IP/天);--smoke零配额链路自检;bs_fundamentals._bs_eod_running补注释:旧任务名已删探测必失败fail-open=设计行为勿修复(探新名会探到父任务永久让路);+15测试(顺序铁律/leg2无条件/退出码聚合9组合/smoke零子进程) [vps]
This commit is contained in:
@@ -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())
|
||||
@@ -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
|
||||
@@ -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(
|
||||
|
||||
@@ -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}
|
||||
Reference in New Issue
Block a user