feat(data): P1+P2+P3季频财务+业绩报告——bs_fundamentals新脚本(sanguo-bs-fund schtask 23:05,VPS)——①P2回灌:profit利润+dupont杜邦2015Q1+按季分片,state文件推进日预算自适应续跑(约2季/天≈3周自动跑完;预算不够整季不拉半季,半段落库也幂等)②P1增量:回灌完成后拉当季+上一季(兜住8-31晚披季报)③P3业绩报告:forecast预告2003+/express快报,每股一次调用与区间无关→月首周日全区间恒定2N query幂等拉齐(--reports-now冒烟)④落parquet data/fundamentals_baostock/保留pubDate列(读侧pubDate<=date过滤防前视,同fundamentals-lookahead-bias-fix模式);去重(code,pubDate,statDate)restated后写者胜,键不足2列退全列绝不按单列丢历史⑤预算铁律(48000/天):bs_eod~11k+本脚本DAILY_CAP=30000串行=≤41k,月首周日叠reports 44.3k<48k余3.7k;开跑前schtasks探sanguo-bs-eod(极端尾22:41)在跑则exit0次日幂等补——同IP永不双连接⑥复用bs_eod四件套(login_with_retry/relogin/_with_timeout/全A清单)+周期relogin+失败relogin兜底;字段名读rs.fields不硬编码;+14测试(季度边界/parquet幂等+防呆/state损坏重头/动态字段+计数/回灌2季cap推进/state超前零拉/增量当季+上季/报告月首周日窗口/守卫文本/main跳过+exit2);data_platform 140绿 [vps]

2026-08-20 09:53:33 +08:00
parent dbb5ed5aa8
commit d0a538ced3
2 changed files with 592 additions and 0 deletions
@@ -0,0 +1,347 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""bs_fundamentals.py — sanguo-bs-fund (VPS schtask 23:05): baostock 季频财务+业绩报告.
+
+2026-08-20 数据补全 P1+P2+P3:
+- P2 历史回灌: profit(利润)+dupont(杜邦) 2015Q1+ 按季分片, 日预算内自适应续跑
+  (state 文件推进, 幂等), 约 2 季/天 × ~46 季 ≈ 3 周自动跑完
+- P1 增量: 回灌完成后拉当季+上一季(兜住晚披季报), 2 表 × 2 季 × 全 A ≈ 22k query
+- P3 业绩报告: forecast(业绩预告 2003+)/express(业绩快报 2006+), 每股一次调用与
+  区间无关 → 月首周日全区间恒定 2×N query 幂等拉齐
+- 落 parquet data/fundamentals_baostock/{table}.parquet, 保留 pubDate 列
+  (读侧按 pubDate<=date 过滤防前视偏差, 同 fundamentals-lookahead-bias-fix 模式)
+
+预算(单 IP 48000/天, 用户铁律"每天限额不要超"):
+    bs_eod ~11k(18:05) + 本脚本 DAILY_CAP=30000(23:05 串行) = ≤41k;
+    月首周日叠 reports 2×N ≈ 11k → 44.3k < 48k, 余 3.7k。
+    开跑前探 sanguo-bs-eod 是否仍在跑(18:05+4.6h 极端尾=22:41), 在跑则 exit 0
+    次日幂等补 —— 同 IP 永不双连接(封禁红线是并发不是重试)。
+
+退出码: 0=完成或当日预算耗尽待续跑或主动跳过; 1=致命; 2=登录失败
+"""
+import argparse
+import datetime as dt
+import json
+import logging
+import os
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+import baostock as bs
+import pandas as pd
+
+from bs_eod import (  # noqa: E402 — 复用同一套健壮性封装(登录重试/重连/超时/全A清单)
+    RELOGIN_EVERY,
+    _with_timeout,
+    fetch_all_stocks_with_timeout,
+    login_with_retry,
+    relogin,
+)
+
+BASE = Path(r"C:\sanguo_vnpy_v2")
+OUT_DIR = Path(os.environ.get("BS_FUND_DIR", str(BASE / "data" / "fundamentals_baostock")))
+BACKFILL_START_YEAR = 2015            # P2 范围起点
+DAILY_CAP = int(os.environ.get("BS_FUND_DAILY_CAP", "30000"))  # 本脚本日预算
+QUERY_COUNT = 0                       # 本脚本自己的 query 计数(粗粒度, cap 余量足)
+BS_INTERVAL = 0.3
+REPORT_START = "2003-01-01"           # 业绩预告最早 2003(快报 2006, 统一起点幂等无差)
+QUARTER_TABLES = ("profit", "dupont")     # P1/P2 季频两表
+REPORT_TABLES = ("forecast", "express")   # P3 业绩预告/快报
+
+logging.basicConfig(level=logging.INFO,
+                    format="%(asctime)s %(levelname)s %(message)s",
+                    handlers=[logging.StreamHandler(sys.stdout)])
+log = logging.getLogger(__name__)
+
+# 四张表统一走 fetch_table: API 形参不同(季频 year/quarter vs 报告 start/end),
+# lambda 闭包归一; 字段名读 rs.fields 不硬编码(防 API 字段变动)。
+API = {
+    "profit": lambda code, year, quarter, start, end:
+        bs.query_profit_data(code=code, year=year, quarter=quarter),
+    "dupont": lambda code, year, quarter, start, end:
+        bs.query_dupont_data(code=code, year=year, quarter=quarter),
+    "forecast": lambda code, year, quarter, start, end:
+        bs.query_forecast_data(code=code, start_date=start, end_date=end),
+    "express": lambda code, year, quarter, start, end:
+        bs.query_express_data(code=code, start_date=start, end_date=end),
+}
+
+
+# ======================== 季度工具 ========================
+
+def prev_quarter(y, q):
+    return (y - 1, 4) if q == 1 else (y, q - 1)
+
+
+def next_quarter_of(y, q):
+    return (y + 1, 1) if q == 4 else (y, q + 1)
+
+
+def quarter_of(d):
+    return (d.year, (d.month - 1) // 3 + 1)
+
+
+def quarter_list(start_year, end_yq):
+    """[(y,q)] 升序: start_year Q1 → end_yq(含)."""
+    ey, eq = end_yq
+    return [(y, q) for y in range(start_year, ey + 1) for q in (1, 2, 3, 4)
+            if (y, q) <= (ey, eq)]
+
+
+def quarter_to_str(yq):
+    return f"{yq[0]}Q{yq[1]}"
+
+
+def quarter_from_str(s):
+    y, q = s.split("Q")
+    return (int(y), int(q))
+
+
+# ======================== 落库 parquet(幂等) ========================
+
+def _state_path():
+    return OUT_DIR / "backfill_state.json"
+
+
+def load_state():
+    if _state_path().exists():
+        try:
+            return json.loads(_state_path().read_text(encoding="utf-8"))
+        except Exception:
+            log.warning("state 文件损坏, 回灌从头(parquet 幂等无伤)")
+    return {"next_quarter": f"{BACKFILL_START_YEAR}Q1"}
+
+
+def save_state(state):
+    OUT_DIR.mkdir(parents=True, exist_ok=True)
+    _state_path().write_text(json.dumps(state, ensure_ascii=False, indent=2),
+                             encoding="utf-8")
+
+
+def append_parquet(table, new_rows):
+    """rows -> OUT_DIR/{table}.parquet, 按 (code,pubDate,statDate) 去重幂等追加。
+
+    restated(同报告期重披)后写者胜; 去重键不足 2 列时退全列去重 —— 绝不按单列
+    (如只剩 code)去重丢历史。
+    """
+    if not new_rows:
+        return 0
+    df = pd.DataFrame(new_rows)
+    p = OUT_DIR / f"{table}.parquet"
+    if p.exists():
+        try:
+            df = pd.concat([pd.read_parquet(p), df])
+        except Exception:
+            log.warning("%s 旧 parquet 读取失败, 本次覆盖写(数据可幂等重拉)", table)
+    keys = [c for c in ("code", "pubDate", "statDate") if c in df.columns]
+    if len(keys) >= 2:
+        df = df.drop_duplicates(subset=keys, keep="last")
+    else:
+        df = df.drop_duplicates(keep="last")
+    sort_keys = [c for c in ("code", "pubDate") if c in df.columns] or list(df.columns)
+    df.sort_values(sort_keys).to_parquet(p, index=False)
+    return len(df)
+
+
+# ======================== 拉取 ========================
+
+def fetch_table(table, bs_code, year=None, quarter=None, start=None, end=None,
+                timeout=60):
+    """拉一张表, 返 list[dict](字段名读 rs.fields 不硬编码), 计 QUERY_COUNT。"""
+    global QUERY_COUNT
+    QUERY_COUNT += 1
+    rs = _with_timeout(lambda: API[table](bs_code, year, quarter, start, end),
+                       timeout=timeout)
+    if rs.error_code != "0":
+        raise RuntimeError(f"{table} {bs_code}: {rs.error_code} {rs.error_msg}")
+    fields = list(rs.fields)
+    rows = []
+    while rs.next():
+        rows.append(dict(zip(fields, rs.get_row_data())))
+    return rows
+
+
+def _sweep_stocks(stocks, fetch_one, label):
+    """通用扫全 A: 逐股 fetch_one(bs_code)->rows, 处理日预算/relogin/间隔。
+
+    返 (rows, failed, completed): completed=False 表示日预算中断(半段数据也有效,
+    幂等, 下次重拉该批去重)。
+    """
+    rows_all = []
+    failed = 0
+    completed = True
+    t0 = time.time()
+    for i, (code, prefix) in enumerate(stocks):
+        if QUERY_COUNT >= DAILY_CAP:
+            log.warning("[%s] query %d 达日预算 %d, 今日到此为止",
+                        label, QUERY_COUNT, DAILY_CAP)
+            completed = False
+            break
+        bs_code = f"{prefix}.{code}"
+        try:
+            rows_all.extend(fetch_one(bs_code))
+        except Exception as e:
+            failed += 1
+            if failed <= 5 or failed % 100 == 0:
+                log.warning("[%s] %s err: %s", label, code, e)
+            if not relogin():
+                log.error("[%s] relogin 失败, 跳过 %s", label, code)
+        if (i + 1) % RELOGIN_EVERY == 0:
+            log.info("[%s] 进度 %d/%d failed=%d q=%d (%.0fs)", label, i + 1,
+                     len(stocks), failed, QUERY_COUNT, time.time() - t0)
+            if not relogin():
+                log.warning("[%s] 周期 relogin 失败, 继续(下次失败被动兜底)", label)
+        if i < len(stocks) - 1:
+            time.sleep(BS_INTERVAL)
+    return rows_all, failed, completed
+
+
+# ======================== 三阶段 ========================
+
+def run_quarter_backfill(stocks, today):
+    """P2: state.next_quarter → 上一季, 按季分片; 预算不够整季则留给明日。
+
+    返 "done"(回灌完成) / "capped"(今日预算耗尽, 明日续)。
+    """
+    state = load_state()
+    yq = quarter_from_str(state["next_quarter"])
+    target = prev_quarter(*quarter_of(today))
+    if yq > target:
+        return "done"
+    log.info("[BACKFILL] %s → %s, 每季 %d query(2表×全A)",
+             quarter_to_str(yq), quarter_to_str(target), 2 * len(stocks))
+    while yq <= target:
+        if QUERY_COUNT > 0 and QUERY_COUNT + 2 * len(stocks) > DAILY_CAP:
+            log.info("[BACKFILL] 预算余 %d 不够整季, 留明日",
+                     DAILY_CAP - QUERY_COUNT)
+            return "capped"
+        for table in QUARTER_TABLES:
+            y, q = yq
+            rows, failed, completed = _sweep_stocks(
+                stocks,
+                lambda c, _t=table, _y=y, _q=q: fetch_table(
+                    _t, c, year=_y, quarter=_q),
+                f"{table}-{quarter_to_str(yq)}")
+            append_parquet(table, rows)
+            if not completed:
+                log.warning("[BACKFILL] %s 半段已落库(幂等), 明日重拉补齐",
+                            quarter_to_str(yq))
+                return "capped"
+        log.info("[BACKFILL] %s 完成 (profit+dupont)", quarter_to_str(yq))
+        state["next_quarter"] = quarter_to_str(next_quarter_of(*yq))
+        save_state(state)
+        yq = next_quarter_of(*yq)
+    return "done"
+
+
+def run_quarter_incremental(stocks, today):
+    """P1: 拉当季+上一季(上一季兜住 8-31 晚披年报/季报), 幂等。"""
+    cur = quarter_of(today)
+    prev = prev_quarter(*cur)
+    log.info("[INC] %s+%s profit+dupont", quarter_to_str(prev), quarter_to_str(cur))
+    for y, q in (prev, cur):
+        for table in QUARTER_TABLES:
+            rows, _, _ = _sweep_stocks(
+                stocks, lambda c, _t=table, _y=y, _q=q: fetch_table(
+                    _t, c, year=_y, quarter=_q),
+                f"inc-{table}-{quarter_to_str((y, q))}")
+            append_parquet(table, rows)
+
+
+def run_reports(stocks, today, force=False):
+    """P3: forecast/express 全区间(每股一次恒定 query), 月首周日跑。
+
+    月频足够(报告类无实时消费者); force=True 供 --reports-now 冒烟/补跑。
+    """
+    if not force and not (today.weekday() == 6 and today.day <= 7):
+        return
+    end = today.strftime("%Y-%m-%d")
+    log.info("[REPORT] forecast+express 全区间 %s~%s", REPORT_START, end)
+    for table in REPORT_TABLES:
+        rows, _, _ = _sweep_stocks(
+            stocks, lambda c, _t=table: fetch_table(
+                _t, c, start=REPORT_START, end=end), table)
+        append_parquet(table, rows)
+
+
+# ======================== bs_eod 在跑守卫 ========================
+
+def _is_running_text(text):
+    """schtasks /query 输出判断 Running(schtasks 可能英/中输出, 两词都认)."""
+    return ("Running" in text) or ("运行中" in text)
+
+
+def _bs_eod_running():
+    """sanguo-bs-eod schtask 是否仍在跑(同 IP 双 baostock 连接红线)。
+
+    探测失败按未在跑处理(fail-open)—— 23:05 距 bs_eod 极端尾 22:41 有 20min 余量。
+    """
+    try:
+        out = subprocess.run(
+            ["schtasks", "/query", "/tn", "sanguo-bs-eod", "/fo", "LIST"],
+            capture_output=True, timeout=30).stdout.decode("gbk", "ignore")
+        return _is_running_text(out)
+    except Exception as e:
+        log.warning("schtasks 探测失败(按未在跑继续): %s", e)
+        return False
+
+
+# ======================== main ========================
+
+def main():
+    global QUERY_COUNT
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--limit", type=int, default=0, help="限股数(冒烟用)")
+    ap.add_argument("--backfill-only", action="store_true",
+                    help="只跑回灌段(增量/报告跳过)")
+    ap.add_argument("--reports-now", action="store_true",
+                    help="无视月首周日限制立即拉报告(冒烟/补跑)")
+    args = ap.parse_args()
+    today = dt.date.today()
+    log.info("bs_fund start cap=%d date=%s", DAILY_CAP, today)
+
+    if _bs_eod_running():
+        log.info("[SKIP] sanguo-bs-eod 仍在跑, 今日让路(exit 0, 次日幂等补)")
+        sys.exit(0)
+
+    if not login_with_retry():
+        log.error("[SKIP] 登录失败, exit 2")
+        sys.exit(2)
+
+    try:
+        stocks = fetch_all_stocks_with_timeout()
+    except Exception as e:
+        log.error("[FATAL] fetch_all: %s", e)
+        sys.exit(1)
+    QUERY_COUNT = 2  # login 探针 + 全 A 清单(本脚本计数, 粗粒度即够)
+    log.info("全 A 含退市: %d 只", len(stocks))
+    if args.limit:
+        stocks = stocks[:args.limit]
+    OUT_DIR.mkdir(parents=True, exist_ok=True)
+
+    status = run_quarter_backfill(stocks, today)
+    if status == "capped":
+        log.info("[DONE-cap] 回灌今日预算耗尽, 明日续跑 query=%d", QUERY_COUNT)
+        _logout()
+        sys.exit(0)
+
+    if not args.backfill_only:
+        run_quarter_incremental(stocks, today)
+        run_reports(stocks, today, force=args.reports_now)
+
+    log.info("[DONE] query=%d", QUERY_COUNT)
+    _logout()
+    sys.exit(0)
+
+
+def _logout():
+    try:
+        bs.logout()
+    except Exception:
+        pass
+
+
+if __name__ == "__main__":
+    main()