# -*- coding: utf-8 -*- """static 三表事件驱动补洗（spec §20.11.7，2026-09-20）。 定位：新鲜度目标本体——dmsk 三表域 NOTICE_DATE 雷达命中股 → 逐股 F10(emweb) 拉近 N 期全列 → upsert 重写 per-stock parquet（static 树自身披露次日新鲜）； 月度全量洗库（VPS ak-monthly）保留兜底扫漏；变更文件落 manifest，晚间 push_vps_mirror_daily.sh 按 manifest 推 VPS（单点采集+镜像分发）。 部署形态：挂 run_dmsk.sh daily lane 之后（一次性容器 /app:ro + 写 /volume1/stock/sanguo_vnpy_v2/data/static；NAS 容器无 akshare → 本脚本自带 emweb 薄客户端，akshare stock_*_sheet_by_report_em 的最小移植）。 雷达语义： - dmsk 近 PARTITION_LOOKBACK 个 dt= 分区 × NOTICE_DATE>=T-RADAR_WINDOW 的行 → (code, domain, REPORT_DATE)→notice；与 static 该股文件**同期**值比对 （per-REPORT_DATE 轴，09-20 review 精化——期级新披露/更正即使 notice 旧于 该股全局 max 亦命中），任一期 dmsk 更新 → 候选。static 缺文件的 code 跳过 （dmsk-only 覆盖段=§19.9 边界，不为北交/老退市造文件）；static 文件损坏 →全期命中→重写自愈。 - 周日 sync_parquet.sh 全量覆盖式回冲=自愈：回冲后 static notice 回退 → 雷达 重新命中重洗（仅 RADAR_WINDOW 窗内；更旧回退归月度全量洗兜底）。 - pending 队列：拉取失败/F10 落库滞后（stale：写入后 max NOTICE_DATE 仍< 雷达值）次日重试，超 PENDING_MAX_AGE 天丢弃记日志。 写纪律（洗库族惯例）：F10 返回零行不写不标（返 empty 进 pending 重试）； tmp+os.replace 原子写；upsert=按 REPORT_DATE 整期替换（重述行天然胜出）， 未命中期原样保留；三个日期列统一 'YYYY-MM-DD' 字符串。 rc：0=完赛（含限额耗尽）；1=存在硬失败（优先）；3=墙钟停（--until，未处理 候选留待次日雷达/pending 自愈）。 """ import argparse import glob import json import os import random import re import sys import time from datetime import datetime, timedelta import pandas as pd EMWEB = "https://emweb.securities.eastmoney.com/PC_HSF10/NewFinanceAnalysis" DOMAIN_ENDPOINT = {"balance": "zcfzb", "income": "lrb", "cashflow": "xjllb"} DOMAINS = ("income", "balance", "cashflow") EMWEB_MIN_INTERVAL = 1.0 # 东财 F10 web 族 1.0s 节流（akshare em 族纪律） PAGE_DATES = 5 # AjaxNew 每发 5 个报告期（akshare 原样） RADAR_WINDOW_DAYS = 14 # 雷达 NOTICE_DATE 窗（覆盖周末+pending 重试） PARTITION_LOOKBACK = 8 # dmsk 近 8 个 dt= 分区（2 年）足够容纳新披露 PENDING_MAX_AGE_DAYS = 30 DATE_COLS = ("REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE") _BAD_DATES = ("", "NaT", "nan", "None", "none") _STATIC_ENV = "SANGUO_STATIC_ROOT" _DMSK_ENV = "SANGUO_FUND_DMSK_ROOT" _STATE_ENV = "SANGUO_STATIC_OVERLAY_STATE" _DEFAULTS = { "static": (r"C:\sanguo_vnpy_v2\data\static" if os.name == "nt" else "/volume1/stock/sanguo_vnpy_v2/data/static"), "dmsk": (r"C:\sanguo_vnpy_v2\data\fundamentals\dmsk" if os.name == "nt" else "/volume1/stock/fundamentals/dmsk"), "state": (r"C:\sanguo_vnpy_v2\data_backup\static_overlay" if os.name == "nt" else "/volume1/stock/sanguo_vnpy_v2/data_backup/static_overlay"), } def _norm_date_series(s: pd.Series) -> pd.Series: return s.astype(str).str[:10] def _valid_notice(s: pd.Series) -> pd.Series: v = _norm_date_series(s) return v[~v.isin(_BAD_DATES) & (v.str.len() == 10)] # ============ emweb F10 薄客户端（akshare 最小移植，fetcher 可注入测试） ============ class EmwebF10Client: """三表按报告期客户端。fetcher(url, params)->json dict / html str 可注入。""" def __init__(self, min_interval=EMWEB_MIN_INTERVAL, fetcher=None, timeout=20, retries=3, retry_sleep=5.0): self._last = [0.0] self.min_interval = min_interval self.timeout = timeout self.retries = retries self.retry_sleep = retry_sleep self._ctype_cache = {} if fetcher is None: import requests self._session = requests.Session() self._session.headers.update({"User-Agent": "Mozilla/5.0"}) self._fetch = self._http_get else: self._session = None self._fetch = fetcher def _throttle(self): wait = self.min_interval - (time.time() - self._last[0]) if wait > 0: time.sleep(wait + random.uniform(0.05, 0.3)) self._last[0] = time.time() def _http_get(self, url, params): r = self._session.get(url, params=params, timeout=self.timeout) r.raise_for_status() return r.json() if url.endswith("AjaxNew") else r.text def _get(self, url, params): err = None for _ in range(self.retries): self._throttle() try: return self._fetch(url, params) except Exception as e: # 网络/JSON 解析双失败统一重试 err = e time.sleep(self.retry_sleep) raise RuntimeError("emweb get fail %s %s: %s" % (url, params, err)) def company_type(self, code_prefixed: str) -> str: if code_prefixed in self._ctype_cache: return self._ctype_cache[code_prefixed] html = self._get(EMWEB + "/Index", {"type": "web", "code": code_prefixed.lower()}) m = (re.search(r'id="hidctype"[^>]*value="([^"]+)"', html) or re.search(r'value="([^"]+)"[^>]*id="hidctype"', html)) if not m: raise RuntimeError("hidctype not found for %s" % code_prefixed) self._ctype_cache[code_prefixed] = m.group(1) return m.group(1) def report_dates(self, domain: str, code_prefixed: str) -> list: ep = DOMAIN_ENDPOINT[domain] data = self._get("%s/%sDateAjaxNew" % (EMWEB, ep), {"companyType": self.company_type(code_prefixed), "reportDateType": "0", "code": code_prefixed}) rows = (data or {}).get("data") or [] return sorted({str(r.get("REPORT_DATE"))[:10] for r in rows if r.get("REPORT_DATE")}) def fetch_recent(self, domain: str, code_prefixed: str, recent_n: int) -> pd.DataFrame: """拉近 recent_n 期（日期降序取前 N，PAGE_DATES 期/发分页拼接）。""" ep = DOMAIN_ENDPOINT[domain] dates = sorted(self.report_dates(domain, code_prefixed), reverse=True)[:recent_n] frames = [] for i in range(0, len(dates), PAGE_DATES): chunk = dates[i:i + PAGE_DATES] data = self._get("%s/%sAjaxNew" % (EMWEB, ep), {"companyType": self.company_type(code_prefixed), "reportDateType": "0", "reportType": "1", "dates": ",".join(chunk), "code": code_prefixed}) rows = (data or {}).get("data") if not rows: continue frames.append(pd.DataFrame(rows)) if not frames: return pd.DataFrame() return pd.concat(frames, ignore_index=True) # ============ 雷达 ============ def scan_radar(dmsk_root: str, static_root: str, domains, window_days: int = RADAR_WINDOW_DAYS): """→ (candidates {(code, domain): dmsk_max_notice}, skipped [(code,domain)])。 比对轴=per-(code, REPORT_DATE)（09-20 factor review MINOR-2 精化）：dmsk 侧 期级 notice 与 static 同期值比对——期级新披露/更正即使 notice 旧于该股全局 max 亦命中（分区名 dt= 即 REPORT_DATE，§20.11.3 契约=分区键同值）。static 文件损坏=cur_map 空→全期命中→重写自愈。candidates 值=该股窗内 dmsk notice 最大值（供 wash_unit stale 校验）。 """ cutoff = (datetime.now() - timedelta(days=window_days)).strftime("%Y-%m-%d") raw = {} # (code, domain, report_date) -> notice for domain in domains: parts = sorted(glob.glob(os.path.join(dmsk_root, domain, "dt=*"))) for pdir in parts[-PARTITION_LOOKBACK:]: rd = os.path.basename(pdir)[3:] for f in glob.glob(os.path.join(pdir, "*.parquet")): try: df = pd.read_parquet( f, columns=["SECURITY_CODE", "NOTICE_DATE"]) except Exception: continue notices = _norm_date_series(df["NOTICE_DATE"]) hit = df[~notices.isin(_BAD_DATES) & (notices >= cutoff).to_numpy() & (notices.str.len() == 10).to_numpy()] for code, notice in zip(hit["SECURITY_CODE"], notices[hit.index]): key = (str(code), domain, rd) if notice > raw.get(key, ""): raw[key] = notice by_stock = {} for (code, domain, rd), notice in raw.items(): by_stock.setdefault((code, domain), {})[rd] = notice candidates, skipped = {}, [] for (code, domain), rd_notices in sorted(by_stock.items()): files = glob.glob(os.path.join( static_root, domain, "%s.*_%s.parquet" % (code, domain))) if not files: skipped.append((code, domain)) continue try: cur = pd.read_parquet( files[0], columns=["REPORT_DATE", "NOTICE_DATE"]) except Exception: cur = None cur_map = {} if cur is not None and len(cur): rds = _norm_date_series(cur["REPORT_DATE"]) nts = _norm_date_series(cur["NOTICE_DATE"]) ok = (~rds.isin(_BAD_DATES) & (rds.str.len() == 10) & ~nts.isin(_BAD_DATES) & (nts.str.len() == 10)) for r, n in zip(rds[ok], nts[ok]): if n > cur_map.get(r, ""): cur_map[r] = n trig, stock_max = False, "" for rd, notice in rd_notices.items(): if notice > stock_max: stock_max = notice if notice > cur_map.get(rd, ""): trig = True if trig: candidates[(code, domain)] = stock_max return candidates, skipped # ============ pending / manifest ============ def load_pending(state_dir: str) -> dict: pend = os.path.join(state_dir, "pending.json") if not os.path.exists(pend): return {"units": []} try: with open(pend, encoding="utf-8") as f: return json.load(f) except Exception: return {"units": []} def save_pending(state_dir: str, pending: dict): os.makedirs(state_dir, exist_ok=True) with open(os.path.join(state_dir, "pending.json"), "w", encoding="utf-8") as f: json.dump(pending, f, ensure_ascii=False, indent=1) def append_manifest(state_dir: str, domain: str, filename: str): """变更清单：manifests/YYYYMMDD.list 每行 '/'（日内幂等）。""" mani = os.path.join(state_dir, "manifests") os.makedirs(mani, exist_ok=True) path = os.path.join(mani, datetime.now().strftime("%Y%m%d") + ".list") line = "%s/%s" % (domain, filename) existing = set() if os.path.exists(path): with open(path, encoding="utf-8") as f: existing = {x.strip() for x in f if x.strip()} if line in existing: return with open(path, "a", encoding="utf-8") as f: f.write(line + "\n") # ============ upsert 与单股洗 ============ def upsert_periods(existing: pd.DataFrame, fetched: pd.DataFrame) -> pd.DataFrame: """按 REPORT_DATE 整期替换（fetched 胜出=重述天然胜），未命中期原样保留。 出口三个日期列统一 'YYYY-MM-DD' 字符串（旧行一并归一，月度全量洗重写 亦同向；消费方 fund_pit/因子 loader 均按前缀归一读，两种历史形态兼容）。 """ if fetched is None or fetched.empty: return existing fetched = fetched.copy() for c in DATE_COLS: if c in fetched.columns: fetched[c] = _norm_date_series(fetched[c]) if existing is None or existing.empty: out = fetched else: hit = set(_norm_date_series(fetched["REPORT_DATE"])) keep = existing[~_norm_date_series( existing["REPORT_DATE"]).isin(hit)] out = pd.concat([keep, fetched], ignore_index=True) for c in DATE_COLS: if c in out.columns: out[c] = _norm_date_series(out[c]) return out def wash_unit(client: EmwebF10Client, static_root: str, state_dir: str, domain: str, code: str, recent_n: int, radar_notice: str = "") -> str: """单 (code, domain)：拉→upsert→原子写→manifest。返 ok/stale/empty。 stale=写入成功但 max NOTICE_DATE 仍 < 雷达值（F10 落库滞后，pending 次日 重试）；网络失败向上 raise 由调用方记 pending。 """ files = glob.glob(os.path.join( static_root, domain, "%s.*_%s.parquet" % (code, domain))) if not files: return "no_static_file" path = files[0] code_prefixed = ("SH" if code.startswith(("6", "9")) else "SZ") + code fetched = client.fetch_recent(domain, code_prefixed, recent_n) if fetched.empty: return "empty" try: existing = pd.read_parquet(path) except Exception: existing = pd.DataFrame() out = upsert_periods(existing, fetched) tmp = path + ".wash_tmp" out.to_parquet(tmp, index=False) os.replace(tmp, path) append_manifest(state_dir, domain, os.path.basename(path)) if "NOTICE_DATE" not in out.columns: return "ok" notices = _valid_notice(out["NOTICE_DATE"]) if not len(notices) or not radar_notice: return "ok" return "ok" if notices.max() >= radar_notice else "stale" # ============ 主流程 ============ def _wall_deadline(until: str) -> datetime: """--until HH:MM 当日墙钟；已过期=立即停（不滚明天——迟到的手动跑 不许越过当日窗口继续拉，corpus 语义同族）。""" h, m = until.split(":") now = datetime.now() d = now.replace(hour=int(h), minute=int(m), second=0, microsecond=0) return d if d >= now else now def run(args, client=None) -> int: static_root = (os.environ.get(_STATIC_ENV) or args.static_root or _DEFAULTS["static"]) dmsk_root = (os.environ.get(_DMSK_ENV) or args.dmsk_root or _DEFAULTS["dmsk"]) state_dir = (os.environ.get(_STATE_ENV) or args.state_dir or _DEFAULTS["state"]) domains = [d for d in args.domains.split(",") if d] deadline = _wall_deadline(args.until) candidates, skipped = scan_radar(dmsk_root, static_root, domains, args.radar_window) print("radar: candidates=%d dmsk_only_skipped=%d" % ( len(candidates), len(skipped))) pending = load_pending(state_dir) today = datetime.now().strftime("%Y-%m-%d") alive, pend_keys = [], set() for u in pending["units"]: age = (datetime.now() - datetime.strptime( u["first_seen"], "%Y-%m-%d")).days if age > PENDING_MAX_AGE_DAYS: print("pending expire: %s/%s (%dd)" % (u["code"], u["domain"], age)) continue alive.append(u) pend_keys.add((u["code"], u["domain"])) worklist = sorted(set(candidates) | pend_keys) if args.limit: worklist = worklist[:args.limit] print("worklist=%d (pending_carry=%d)" % (len(worklist), len(pend_keys))) client = client or EmwebF10Client() hard_fail, stale_units, done_ok = [], [], set() washed = 0 rc_wall = False for i, (code, domain) in enumerate(worklist): if datetime.now() >= deadline: print("wallclock stop at %d/%d" % (i, len(worklist))) rc_wall = True break radar_notice = candidates.get((code, domain), "") if args.dry_run: print("[dry] %s/%s radar_notice=%s" % (code, domain, radar_notice or "(pending)")) continue try: status = wash_unit(client, static_root, state_dir, domain, code, args.recent_periods, radar_notice) print("washed %s/%s -> %s" % (code, domain, status)) if status == "ok": washed += 1 done_ok.add((code, domain)) else: # stale/empty=次日重试；no_static_file 不该出现在 worklist stale_units.append({"code": code, "domain": domain}) except Exception as e: print("FAIL %s/%s: %s" % (code, domain, e)) hard_fail.append((code, domain)) # pending 落盘：ok 出队；未触达（墙钟停/限额）沿用；失败/stale 续留 if not args.dry_run: new_pend = {u["code"] + "/" + u["domain"]: u for u in alive if (u["code"], u["domain"]) not in done_ok} for u in stale_units: new_pend.setdefault(u["code"] + "/" + u["domain"], {"code": u["code"], "domain": u["domain"], "first_seen": today}) for code, domain in hard_fail: new_pend.setdefault(code + "/" + domain, {"code": code, "domain": domain, "first_seen": today}) save_pending(state_dir, {"units": sorted( new_pend.values(), key=lambda u: (u["code"], u["domain"]))}) print("summary: washed=%d fail=%d retry_pending=%d pending_total=%d" % ( washed, len(hard_fail), len(stale_units), 0 if args.dry_run else len(new_pend))) if hard_fail: return 1 if rc_wall: return 3 return 0 def main(): ap = argparse.ArgumentParser(description="static 三表事件驱动补洗 (§20.11.7)") ap.add_argument("--until", default="15:00", help="墙钟停 HH:MM") ap.add_argument("--limit", type=int, default=0, help="本次最多洗 N 单元(0=不限)") ap.add_argument("--recent-periods", type=int, default=12, help="每股域拉近 N 期（默认 12=3 年，深史归月度全量洗）") ap.add_argument("--radar-window", type=int, default=RADAR_WINDOW_DAYS) ap.add_argument("--domains", default=",".join(DOMAINS)) ap.add_argument("--static-root", default=None) ap.add_argument("--dmsk-root", default=None) ap.add_argument("--state-dir", default=None) ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() rc = run(args) print("static_overlay_wash rc=%d" % rc) sys.exit(rc) if __name__ == "__main__": main()