446 lines
19 KiB
Python
446 lines
19 KiB
Python
# -*- 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 每行 '<domain>/<file>'（日内幂等）。"""
|
||
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()
|