fix: market-sync 2h timeout 截断 + 加 daily-check 巡检

work #01 (2026-07-03):昨天 15:30 runall 中 kline_5min 全量 69min 把 2h
窗口吃满,share_snapshot / market_regime 被 systemd SIGTERM 截断;
今天又叠加 worker.py 未长驻,导致 10 个 task missed。

变更:
- bin/systemd/market-sync.service: TimeoutStartSec 7200 → 14400 (4h)
- bin/daily_sync_check.py: 新增每日巡检,读 PG dataset_registry,
  对比 SCHEDULE 判定 ok/missed/failed/stuck/never_run,写 JSON 报告,
  失败时 POST HMAC webhook
- bin/systemd/market-sync-daily-check.{service,timer}: 23:00 daily 触发
  daily_check(替代 session-only 的 /loop 方案)
- config/daily_check_secrets.env.template: webhook 配置模板
- docs/works/2026-07-03-01-market-sync-timeout.md: 完整 work 记录
  (根因 + 中期/长期改进建议)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
gao
2026-07-03 10:48:20 +08:00
parent 7a985dddd5
commit 295177027f
6 changed files with 542 additions and 2 deletions
+408
View File
@@ -0,0 +1,408 @@
"""每日同步任务巡检。
读 PG `dataset_registry`,对比今日预期调度窗口,给出每个 task 的健康状态,
写 `logs/daily_check_YYYYMMDD_HHMMSS.json` 报告,仅在有失败/漏跑时 POST webhook。
用法:
.venv/bin/python bin/daily_sync_check.py # 默认检查"今天"
.venv/bin/python bin/daily_sync_check.py --dry-run # 只生成报告,不 POST
.venv/bin/python bin/daily_sync_check.py --report-only # 完全不读 secrets / 不 POST
.venv/bin/python bin/daily_sync_check.py --date 2026-07-02 # 检查指定日期
为什么是 stand-alone(不放在 app.tasks 下):
- 巡检逻辑独立于 sync 业务,混在 app.tasks 会让注册表臃肿
- 独立脚本便于 cron / systemd timer / /loop 三种触发方式共用
"""
from __future__ import annotations
import argparse
import hashlib
import hmac
import json
import logging
import os
import re
import sys
import time
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Optional
import requests
# ── 路径与日志 ─────────────────────────────────────────────────────────────
PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
LOG_DIR = PROJECT_ROOT / "logs"
SECRETS_FILE = PROJECT_ROOT / "config" / "daily_check_secrets.env"
LOG = logging.getLogger("daily-sync-check")
# ── 预期调度表 ─────────────────────────────────────────────────────────────
# 与 [[market-data-overview]] 保持一致;同步触发点(HH:MM)按 Asia/Shanghai 解释。
# weekend_only=True 表示只在周六/周日期望运行(实际仅 stock_node 周六一次)。
SCHEDULE: dict[str, dict[str, Any]] = {
"stock_basic": {"window_end": "09:30", "weekend_only": False},
"industry_sector": {"window_end": "09:45", "weekend_only": False},
"kline_index": {"window_end": "15:35", "weekend_only": False},
"kline_daily": {"window_end": "15:45", "weekend_only": False},
"kline_5min": {"window_end": "16:05", "weekend_only": False},
"market_regime": {"window_end": "16:20", "weekend_only": False},
"share_snapshot": {"window_end": "16:25", "weekend_only": False},
"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}, # 周六
}
# ── 状态判定 ───────────────────────────────────────────────────────────────
def _parse_ts(s: Optional[str]) -> Optional[datetime]:
"""dataset_registry 时间字段是 '%Y-%m-%d %H:%M:%S' 或 None。"""
if not s:
return None
for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"):
try:
return datetime.strptime(s, fmt)
except ValueError:
continue
return None
def _classify_task(
row: dict[str, Any],
check_date: date,
now: datetime,
) -> dict[str, Any]:
"""判定单个 task 在 check_date 的健康状态。
返回字段:
status: ok | missed | failed | running | never_run | not_expected
reason: 简短中文原因
last_success_at / last_failure_at / last_error / needs_resync: 原值透传
"""
dataset_id = row["dataset_id"]
enabled = (row.get("enabled") or 0) == 1
spec = SCHEDULE.get(dataset_id)
is_weekend = check_date.weekday() >= 5
if not enabled:
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "disabled",
"reason": "enabled=0",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": row.get("status", "idle"),
}
# weekend 期望判定
if spec and spec["weekend_only"] and not is_weekend:
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "not_expected",
"reason": "周末专属任务,今天不期望",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": row.get("status", "idle"),
}
if spec and not spec["weekend_only"] and is_weekend:
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "not_expected",
"reason": "周末不调度",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": row.get("status", "idle"),
}
last_success = _parse_ts(row.get("last_success_at"))
last_failure = _parse_ts(row.get("last_failure_at"))
status_field = row.get("status", "idle")
# 还在跑(卡死检测)
if status_field == "running":
started = _parse_ts(row.get("started_at"))
if started and (now - started) > timedelta(hours=2):
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "stuck",
"reason": f"running > 2h, started_at={row.get('started_at')}",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "running",
"reason": f"status=running, started_at={row.get('started_at')}",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
# 从未跑过
if last_success is None and last_failure is None:
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "never_run",
"reason": "从未同步过",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
# 今日是否有成功?
today_start = datetime.combine(check_date, datetime.min.time())
if last_success and last_success >= today_start:
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "ok",
"reason": f"今日已成功 @ {last_success.strftime('%H:%M:%S')}",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
# 今日失败过?
if last_failure and last_failure >= today_start and (
not last_success or last_failure > last_success
):
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "failed",
"reason": f"今日失败 @ {last_failure.strftime('%H:%M:%S')}: "
f"{(row.get('last_error') or '')[:120]}",
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
# 期望窗口已过但今日没成功
return {
"dataset_id": dataset_id,
"name": row.get("name", ""),
"status": "missed",
"reason": (
f"今日未成功(last_success={last_success.strftime('%Y-%m-%d %H:%M') if last_success else 'never'}"
),
"last_success_at": row.get("last_success_at"),
"last_failure_at": row.get("last_failure_at"),
"last_error": row.get("last_error"),
"needs_resync": row.get("needs_resync", 0),
"status_field": status_field,
}
# ── 报告生成 ───────────────────────────────────────────────────────────────
def build_report(check_date: date) -> dict[str, Any]:
"""生成当日巡检报告。"""
from app.core.db import ops as db_ops # 延迟导入:避免单独跑这个脚本时加载 .env
now = datetime.now()
rows = db_ops.fetch_dataset_registry_rows()
classified = [_classify_task(r, check_date, now) for r in rows]
# 汇总计数
counts: dict[str, int] = {}
for c in classified:
counts[c["status"]] = counts.get(c["status"], 0) + 1
# 告警列表:只挑非 ok / non-expected 的
alerts = [
{"dataset_id": c["dataset_id"], "status": c["status"], "reason": c["reason"]}
for c in classified
if c["status"] in {"missed", "failed", "stuck", "never_run"}
]
overall = "ok" if not alerts else (
"warning" if any(a["status"] in {"missed", "never_run"} for a in alerts) else "error"
)
return {
"schema_version": 1,
"generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
"check_date": check_date.isoformat(),
"overall": overall,
"counts": counts,
"alerts": alerts,
"tasks": classified,
}
def write_report(report: dict[str, Any]) -> Path:
"""把报告写到 logs/daily_check_YYYYMMDD_HHMMSS.json。"""
LOG_DIR.mkdir(parents=True, exist_ok=True)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
path = LOG_DIR / f"daily_check_{ts}.json"
path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
return path
def print_summary(report: dict[str, Any]) -> None:
"""stdout 一眼看懂的汇总。"""
print("=" * 78)
print(f"每日同步巡检 date={report['check_date']} overall={report['overall']}")
print("=" * 78)
print(f"counts: {report['counts']}")
if report["alerts"]:
print(f"\n告警 ({len(report['alerts'])}):")
for a in report["alerts"]:
print(f" - [{a['status']:>9}] {a['dataset_id']:<20} {a['reason']}")
else:
print("\n✅ 所有任务今日已成功")
# 完整列表
print(f"\n全部 {len(report['tasks'])} 个任务:")
for t in report["tasks"]:
icon = {
"ok": "",
"missed": "",
"failed": "",
"stuck": "⚠️",
"running": "🔄",
"never_run": "",
"not_expected": "",
"disabled": "🚫",
}.get(t["status"], "·")
print(f" {icon} [{t['status']:>13}] {t['dataset_id']:<20} {t['reason']}")
# ── Webhook ────────────────────────────────────────────────────────────────
SIG_HEADER_CANDIDATES: list[tuple[str, str]] = [
("X-Hub-Signature-256", "sha256={sig}"),
("X-Webhook-Signature", "{sig}"),
("X-Gitlab-Token", "{secret}"),
]
HTTP_TIMEOUT = 10
def _load_secrets() -> dict[str, str]:
out: dict[str, str] = {}
if not SECRETS_FILE.exists():
return out
try:
for raw in SECRETS_FILE.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
if "=" not in line:
continue
k, v = line.split("=", 1)
k = k.strip()
v = v.strip().strip('"').strip("'")
if k and v:
out[k] = v
except Exception as e:
LOG.error("failed to parse %s: %s", SECRETS_FILE, e)
return out
def post_webhook(url: str, secret: str, payload: dict[str, Any], *, dry_run: bool = False) -> int:
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
sig = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
if dry_run:
print(f"[webhook DRY-RUN] {url}")
print(json.dumps(payload, ensure_ascii=False, indent=2))
return 200
for hdr_name, value_fmt in SIG_HEADER_CANDIDATES:
sig_value = value_fmt.format(sig=sig, secret=secret)
headers = {"Content-Type": "application/json", hdr_name: sig_value}
try:
r = requests.post(url, data=body, headers=headers, timeout=HTTP_TIMEOUT)
if r.status_code in (401, 403):
LOG.info("auth header %s rejected (%s) — trying next", hdr_name, r.status_code)
continue
return r.status_code
except requests.RequestException as e:
LOG.error("POST %s failed with %s: %s", url, hdr_name, e)
return 0
return 0
def maybe_alert(report: dict[str, Any], secrets: dict[str, str], *, dry_run: bool) -> None:
"""只在 overall != ok 时 POST。"""
if report["overall"] == "ok":
LOG.info("全部 ok,跳过 webhook")
return
url = secrets.get("FAILURE_WEBHOOK_URL")
secret = secrets.get("FAILURE_WEBHOOK_SECRET")
if not url or not secret:
LOG.warning("FAILURE_WEBHOOK_URL/SECRET 未配置,跳过 POST")
return
payload = {
"task": "daily_check",
"service": "bin/daily_sync_check.py",
"status": report["overall"],
"check_date": report["check_date"],
"counts": report["counts"],
"alerts": report["alerts"],
"generated_at": report["generated_at"],
}
status = post_webhook(url, secret, payload, dry_run=dry_run)
LOG.info("webhook POST → %s", status)
# ── CLI ────────────────────────────────────────────────────────────────────
def main(argv: Optional[list] = None) -> int:
parser = argparse.ArgumentParser(description="每日同步巡检")
parser.add_argument("--date", help="检查日期 YYYY-MM-DD(默认今天)")
parser.add_argument("--dry-run", action="store_true", help="webhook 走 dry-run(仍写报告)")
parser.add_argument("--report-only", action="store_true", help="不读 secrets、不 POST")
args = parser.parse_args(argv)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(name)s: %(message)s",
datefmt="%Y-%m-%dT%H:%M:%S",
)
if args.date:
check_date = datetime.strptime(args.date, "%Y-%m-%d").date()
else:
check_date = date.today()
report = build_report(check_date)
path = write_report(report)
print_summary(report)
LOG.info("报告已写: %s", path)
if not args.report_only:
secrets = _load_secrets()
maybe_alert(report, secrets, dry_run=args.dry_run)
return 0 if report["overall"] == "ok" else 1
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,31 @@
[Unit]
Description=Market sync daily check — reads dataset_registry, writes report, alerts on failures
Documentation=file:///home/gao/Development/quant_home/market_sync/bin/daily_sync_check.py
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
WorkingDirectory=/home/gao/Development/quant_home/market_sync
User=gao
Group=gao
# 巡检:读 PG dataset_registry,对比今日预期,写 logs/daily_check_*.json
# 失败时 POST webhook 到 FAILURE_WEBHOOK_URLHMAC-SHA256 签名)
ExecStart=/home/gao/Development/quant_home/market_sync/.venv/bin/python /home/gao/Development/quant_home/market_sync/bin/daily_sync_check.py
# 巡检应 < 30s 完成;超时 = PG 不通或代码 bug,不让卡死
TimeoutStartSec=120
# 不要 Restart=oneshot 失败就让 OnFailure= 发通知,别自动重跑)
# 失败时 exit code = 1,可被 OnFailure= 钩住发通知
# 日志走 journaldjournalctl -u market-sync-daily-check -f
StandardOutput=journal
StandardError=journal
SyslogIdentifier=market-sync-daily-check
Environment=PYTHONUNBUFFERED=1
[Install]
WantedBy=multi-user.target
+15
View File
@@ -0,0 +1,15 @@
[Unit]
Description=Schedule market sync daily check — Daily 23:00 Asia/Shanghai
# market-sync-daily-check.service 是这个 timer 的执行单元
# 23:00 触发:周一~五最后一个任务是 longhubang @ 22:00,留 1h buffer
# 周六 23:00 给 stock_node @ 11:30 留 11h+ buffer
[Timer]
# 每天 23:00 触发;Persistent=true 确保关机/错过时下次开机补跑
OnCalendar=*-*-* 23:00:00 Asia/Shanghai
Persistent=true
Unit=market-sync-daily-check.service
# 不要 AccuracySec(默认 1min 漂移够用)
[Install]
WantedBy=timers.target
+4 -2
View File
@@ -13,8 +13,10 @@ Group=gao
# 入口脚本(runall_once.py + structured_watch.sh 二合一)
ExecStart=/home/gao/Development/quant_home/market_sync/bin/market_sync_run.sh
# 硬上限 2h(实际 ~80min留 buffer 给 retry / 慢任务)
TimeoutStartSec=7200
# 硬上限 4h(实际 ~80min + kline_5min 增量 70min, 总 ~2.5h, 留 buffer 给 retry / 慢任务)
# 历史 bug2026-07-02):2h 超时截断,share_snapshot / market_regime 没跑成
# 详见 docs/works/2026-07-03-01-market-sync-timeout.md
TimeoutStartSec=14400
# 不要 Restart=oneshot 失败就让 OnFailure= 发通知,别自动重跑——
# 重跑会撞外部 API 限流窗口,浪费 1.5h)
+6
View File
@@ -0,0 +1,6 @@
# 每日巡检 webhook 配置(HMAC-SHA256 签名)
# 复制为 daily_check_secrets.env 后填入真实值,文件权限 600
# 失败 / 漏跑时调用(必填,否则只在本地报告)
FAILURE_WEBHOOK_URL=
FAILURE_WEBHOOK_SECRET=
@@ -0,0 +1,78 @@
# 2026-07-03 — work #01: market-sync.service 2h timeout 截断同步
## 发现
`bin/daily_sync_check.py --report-only`(周五 10:31)→ 10 个 missed
```
❌ stock_basic / kline_daily / kline_index / kline_5min / tick_trade
❌ moneyflow / industry_sector / sector_features / market_regime / longhubang
✅ share_snapshot (今日 00:03:01 手动跑过)
⏸ stock_node (周末专属任务,今天周五不期望)
```
## 根因
不是今天调度出问题,是**昨天 2026-07-02 15:30 runall 没跑完**
```
15:30:01 启动 runall
15:38:50 stock_basic 完成 (529s)
15:38:58 kline_index 完成 (2s)
15:56:54 kline_daily 完成 (1072s) warning
17:06:09 kline_5min 完成 (4150s=69min) ← 单 task 吃掉一半窗口
17:22:37 industry_sector 完成 (983s) warning (baostock timeout)
17:24:27 sector_features 完成 (105s)
17:24:32 >>> 开始 share_snapshot
17:30:01 systemd SIGTERM 杀进程 ← TimeoutStartSec=7200 (2h) 到期
✗ share_snapshot 没跑
✗ market_regime 没跑
```
后果链:
1. 昨天 kline_5min 全量同步 5207 只耗时 69 min4150s),单 task 吃掉 1/2 窗口
2. industry_sector 又因 baostock connection timed out 浪费 16 min
3. 17:30 systemd timeout 砍掉 share_snapshot + market_regime
4. 今天 09:00 该跑的 stock_basic 没跑 —— 因为 worker.py 没长驻 + market-sync.timer 只在 15:30 触发
5. tick/moneyflow/lhb 几个独立 timer 在 21:05/21:35/22:00 跑过但今天又没补
## 解决
### 短期(本次 commit
- `bin/systemd/market-sync.service`: `TimeoutStartSec``7200` (2h) 提到 `14400` (4h)
- 注释 `实际 ~80min` → 改成 `~80min + kline_5min 增量 70min, 总 ~2.5h, 留 buffer`
-`bin/systemd/market-sync-daily-check.timer` + `.service`
- 每天 23:00 跑 `bin/daily_sync_check.py`,替代 `/loop` 方案的 session-only 限制
### 中期(不在本次 commit,记录备查)
- kline_5min 改成增量模式(当前是全量 5207 只每次),预计从 69min 降到 5min
- 增量条件:`bar_time >= 上次同步时间 - 1 天`(含补漏)
- worker.py 长驻 + 进程内 scheduler 接管盘后 task 调度,与 systemd timer 二选一([[dual-scheduler-overlap]]
- 失败 task 自动 retry 一次(仅 failed,不 retry partial-success / warning
### 长期(架构改进)
- runall 拆成 fast-path / slow-path 两个 service
- fast-path: stock_basic / kline_index / kline_daily / industry_sector / share_snapshot / market_regime~30min
- slow-path: kline_5min 全量(独立 timer @ 周日凌晨,避免挤占交易日窗口)
## 关联
- [[background-watcher-pitfalls]]: "不要让 systemd TimeoutStartSec= 自动收尾"
- 实际情况是 runall 卡在 kline_5minwrapper 的 fast-poll 也没起到作用(runall 在 2h 内确实没退出,systemd 超时是对的)
- [[dual-scheduler-overlap]]: worker.py 没运行 → 完全依赖 systemd timer
- [[market-data-overview]]: 任务耗时分布
## 验收
commit 后验证:
```bash
sudo cp bin/systemd/market-sync.service /etc/systemd/system/
sudo cp bin/systemd/market-sync-daily-check.{service,timer} /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl restart market-sync.timer
systemctl status market-sync-daily-check.timer
# 等今晚 23:00 看 daily_check 报告
```