astock-kg:D案盘前链——beat 挪 06:20/06:35/06:40/06:50 周二~六;

hotspot/transmission 不传日期时扫描日=最新快照日,标签与数据构造对齐

akg-factor-bridge:D案配套——build daily 默认日=基座最新扫描日;
G1 第七节记 D 案拍板与落法
This commit is contained in:
zlt 2026-07-30 13:57:57 +08:00
parent 459028231c
commit d00c1f262a
2 changed files with 31 additions and 3 deletions

View File

@ -270,3 +270,13 @@ probe 三张读数提醒已钉 **07-30 周四 19:00**。
落 data/plan/,终端同步打印。 落 data/plan/,终端同步打印。
**工作哲学07-30 用户令)**:数据渐进式暴露是常态——不等全量抽完、不等干净 **工作哲学07-30 用户令)**:数据渐进式暴露是常态——不等全量抽完、不等干净
截面、不等覆盖完备;缺锚降档不弃用,滞后交采信规则消化。 截面、不等覆盖完备;缺锚降档不弃用,滞后交采信规则消化。
6. **拍点已拍D 案07-30 用户)——整链挪次日盘前。** 落法:基座 beat 改
06:20/06:35 双拍 sync → 06:40 hotspot → 06:50 transmission跑**周二~周六**
(每晨补全上一交易日的完整截面;周六晨补周五,周一盘前读周六产出);
hotspot/transmission 不传日期时**扫描日=最新快照日(数据日)**,标签与数据
由构造对齐,"快照非扫描日"告警只在显式指定日期的历史重扫时才可能触发;
桥 build daily 默认日改为基座最新扫描日(不再取"今天"),桥宿主 cron 改
UTC `10 23 * * 1-5`=CST 周二~六 07:10build+plan 连跑。免责声明按用户令
从计划尾部移除。部署:基座 pull 后 force-recreate beat + worker-interactive
桥 pull + 改宿主 crontab。触发方式后续演进为接口回调XXL-JOB/事件驱动,
用户 07-30 提出——celery 任务与桥 CLI 均可被外部触发,届时只换触发器不换码。

22
run.py
View File

@ -11,7 +11,8 @@
python run.py build akg_heat --mode daily --date 2026-07-24 python run.py build akg_heat --mode daily --date 2026-07-24
python run.py build akg_event --mode history --start 2025-01-01 --end 2026-07-24 python run.py build akg_event --mode history --start 2025-01-01 --end 2026-07-24
daily 模式不给 --date 则取今天start=end=date跑完自动冻结--no-freeze 可关 daily 模式不给 --date 则取基座最新扫描日盘前链口径07-30 D
start=end=date跑完自动冻结--no-freeze 可关
history 模式 --start/--end默认不逐日冻结 history 模式 --start/--end默认不逐日冻结
所有写入幂等删涉及日期区间再插可安全重跑 所有写入幂等删涉及日期区间再插可安全重跑
""" """
@ -22,6 +23,8 @@ import warnings
# pandas 用 DBAPI 连接读 SQL 会 warn无害功能正常——静音保持日志干净 # pandas 用 DBAPI 连接读 SQL 会 warn无害功能正常——静音保持日志干净
warnings.filterwarnings("ignore", message=".*only supports SQLAlchemy.*") warnings.filterwarnings("ignore", message=".*only supports SQLAlchemy.*")
import pandas as pd
import common import common
import config import config
import db import db
@ -107,9 +110,24 @@ def cmd_register():
common.register(code, name, factors.FACTORS[code], ["astock-kg", code.split("_", 1)[1]], desc) common.register(code, name, factors.FACTORS[code], ["astock-kg", code.split("_", 1)[1]], desc)
def _latest_data_day() -> str:
"""daily 不传 --date 时的默认日:基座传导台账最新 scan_date=数据日)。
盘前链07-30 D 在次日早晨补全上一交易日"今天"多半还没有数据
以基座刚完成的扫描日为准两边永远对齐查不到再退回今天"""
try:
df = db.read_pg("SELECT MAX(scan_date) d FROM v_factor_transmission")
v = None if df.empty else df.iloc[0, 0]
if v is not None and not pd.isna(v):
return pd.Timestamp(v).date().isoformat()
except Exception as e: # noqa: BLE001 —— 视图不可达时退回今天,不阻塞构建
print(f" (取基座最新扫描日失败,默认改用今天: {e!r})")
return dt.date.today().isoformat()
def cmd_build(which, mode, start, end, date, do_freeze=True): def cmd_build(which, mode, start, end, date, do_freeze=True):
if mode == "daily": if mode == "daily":
d = date or dt.date.today().isoformat() d = date or _latest_data_day()
start = end = d start = end = d
if not start or not end: if not start or not end:
raise SystemExit("history 模式需要 --start 与 --end") raise SystemExit("history 模式需要 --start 与 --end")