from __future__ import annotations import csv import json import math import re from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from typing import Any import pandas as pd PACKAGE_ROOT = Path(__file__).resolve().parents[1] PROJECT_ROOT = PACKAGE_ROOT.parents[2] RESULT_ROOT = PROJECT_ROOT / "ana-data" / "result" STRICT_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001" READABLE_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-V1-PACKAGE-READABILITY-REPAIR-20260612-001" FULL_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-FULL-2023-2026-20260608-001" DAILY_ROOT = Path(r"E:\quant\a_share_daily_front_20230101_20260508_complete") DAILY_DIR = DAILY_ROOT / "daily" MINUTE_ROOT = Path(r"E:\quant\2023_front_m") RUN_ID = "RUN-ANA-WUJI-FRONTDATA-RULE-AUDIT-20260613-001" def read_csv(path: Path) -> pd.DataFrame: return pd.read_csv(path, dtype=str, keep_default_na=False, encoding="utf-8-sig") def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str] | None = None) -> None: if fieldnames is None: keys: list[str] = [] seen: set[str] = set() for row in rows: for key in row: if key not in seen: keys.append(key) seen.add(key) fieldnames = keys with path.open("w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(rows) def fnum(value: Any) -> float | None: if value is None: return None text = str(value).strip() if text == "": return None try: v = float(text) except ValueError: return None if math.isnan(v): return None return v def bval(value: Any) -> bool | None: text = str(value).strip().lower() if text in {"true", "1", "yes", "y"}: return True if text in {"false", "0", "no", "n"}: return False return None def date_key(value: Any) -> str: text = str(value).strip() if not text: return "" return text.replace("-", "")[:8] def date_display(key: str) -> str: if len(key) == 8: return f"{key[:4]}-{key[4:6]}-{key[6:8]}" return key def symbol_code(symbol: str) -> str: return symbol.split(".")[0] def symbol_market(symbol: str) -> str: if "." in symbol: return symbol.split(".")[-1] code = symbol_code(symbol) if code.startswith("6"): return "SH" if code.startswith(("8", "4", "920")): return "BJ" return "SZ" def minute_path(symbol: str) -> Path: market = symbol_market(symbol) code = symbol_code(symbol) return MINUTE_ROOT / market / f"price_{code}.csv" def daily_path(symbol: str) -> Path: return DAILY_DIR / f"{symbol}.csv" def board_limit_threshold(symbol: str) -> float: code = symbol_code(symbol) market = symbol_market(symbol) if market == "BJ" or code.startswith(("8", "4", "920")): return 29.5 if code.startswith(("300", "301", "688")): return 19.5 return 9.5 DAILY_CACHE: dict[str, pd.DataFrame | None] = {} def get_daily(symbol: str) -> pd.DataFrame | None: if symbol in DAILY_CACHE: return DAILY_CACHE[symbol] path = daily_path(symbol) if not path.exists() or path.stat().st_size == 0: DAILY_CACHE[symbol] = None return None df = pd.read_csv(path, dtype={"trade_date": str}) df["trade_date"] = df["trade_date"].astype(str).str[:8] for col in ["open", "high", "low", "close", "volume", "amount", "preClose", "suspendFlag"]: if col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") df = df.sort_values("trade_date").reset_index(drop=True) DAILY_CACHE[symbol] = df return df def index_for_date(df: pd.DataFrame, key: str) -> int | None: matches = df.index[df["trade_date"] == key].tolist() if not matches: return None return int(matches[0]) def near_equal(a: float | None, b: float | None, rel: float = 1e-6, abs_tol: float = 1e-4) -> bool | None: if a is None or b is None: return None return abs(a - b) <= max(abs_tol, abs(b) * rel) def classify_buy_window(trade_time: str) -> bool: return bool(trade_time and (trade_time < "10:40:00" or trade_time >= "14:40:00")) def build_source_buy_trace(strict_orders: pd.DataFrame, full_orders: pd.DataFrame, selected: pd.DataFrame) -> list[dict[str, Any]]: full_orders_by_id = full_orders.set_index("order_id", drop=False).to_dict("index") full_order_ids = set(full_orders_by_id) selected_by_id = selected.set_index("candidate_id", drop=False).to_dict("index") rows: list[dict[str, Any]] = [] for row in strict_orders.to_dict("records"): if row.get("action") != "BUY" or row.get("source_order_id") not in full_order_ids: continue src = full_orders_by_id.get(row.get("source_order_id", "")) cand = selected_by_id.get(row.get("candidate_id", "")) evidence_rel = row.get("evidence_image_path", "") evidence_path = FULL_ROOT / evidence_rel if evidence_rel else None rows.append( { "strict_order_id": row.get("order_id", ""), "source_order_id": row.get("source_order_id", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "entry_trade_date": row.get("trade_date", ""), "entry_time": row.get("trade_time", ""), "candidate_id": row.get("candidate_id", ""), "source_order_found": True, "candidate_found": bool(cand), "source_evidence_image_path": str(evidence_path) if evidence_path else "", "source_evidence_image_exists": bool(evidence_path and evidence_path.exists()), "normal_buy_time_window_ok": classify_buy_window(row.get("trade_time", "")), "candidate_rank": cand.get("candidate_rank", "") if cand else "", "candidate_rank_top5": (fnum(cand.get("candidate_rank")) <= 5) if cand and fnum(cand.get("candidate_rank")) is not None else "", "market_gate_open_flag": cand.get("market_gate_open_flag", "") if cand else "", "old_up_count": cand.get("up_count", "") if cand else "", "old_up_count_ge3000": (fnum(cand.get("up_count")) >= 3000) if cand and fnum(cand.get("up_count")) is not None else "", "strict_candidate_flag": cand.get("strict_candidate_flag", "") if cand else "", "old_volume_ratio": cand.get("volume_ratio", "") if cand else "", "old_recent_limitup_30_flag": cand.get("recent_limitup_30_flag", "") if cand else "", "old_prev_high_volume_pass_flag": cand.get("prev_high_volume_pass_flag", "") if cand else "", "old_decision_reason_cn": src.get("decision_reason_cn", "") if src else "", } ) return rows def build_rolling_buy_trace(strict_orders: pd.DataFrame, full_orders: pd.DataFrame, rolling_signals: pd.DataFrame) -> list[dict[str, Any]]: full_order_ids = set(full_orders["order_id"].tolist()) rolling_by_id = rolling_signals.set_index("rolling_signal_id", drop=False).to_dict("index") rows: list[dict[str, Any]] = [] for row in strict_orders.to_dict("records"): if row.get("action") != "BUY" or row.get("source_order_id") in full_order_ids: continue rolling = rolling_by_id.get(row.get("source_signal_id", "")) or rolling_by_id.get(row.get("source_order_id", "")) evidence_rel = row.get("evidence_image_path", "") evidence_path = STRICT_ROOT / evidence_rel if evidence_rel else None rows.append( { "strict_order_id": row.get("order_id", ""), "rolling_signal_id": row.get("source_signal_id", "") or row.get("source_order_id", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "rolling_trade_date": row.get("trade_date", ""), "rolling_time": row.get("trade_time", ""), "candidate_id": row.get("candidate_id", ""), "rolling_signal_found": bool(rolling), "rolling_window_1040_1440_ok": bool(row.get("trade_time") and "10:40:00" <= row.get("trade_time") <= "14:40:00"), "evidence_image_path": str(evidence_path) if evidence_path else "", "evidence_image_exists": bool(evidence_path and evidence_path.exists()), "rolling_price": rolling.get("rolling_price", "") if rolling else "", "ma5_close": rolling.get("ma5_close", "") if rolling else "", "near_ma5_pct": rolling.get("near_ma5_pct", "") if rolling else "", "volume_ratio_vs_prev20m": rolling.get("volume_ratio_vs_prev20m", "") if rolling else "", "decision_reason_cn": row.get("decision_reason_cn", ""), } ) return rows def recalc_daily_candidate_rules(source_buy_rows: list[dict[str, Any]], selected: pd.DataFrame) -> list[dict[str, Any]]: selected_by_id = selected.set_index("candidate_id", drop=False).to_dict("index") out: list[dict[str, Any]] = [] for buy in source_buy_rows: cand = selected_by_id.get(buy["candidate_id"]) symbol = buy["symbol"] signal_key = date_key(cand.get("signal_trade_date", "") if cand else "") row_out: dict[str, Any] = { "strict_order_id": buy["strict_order_id"], "case_id": buy["case_id"], "symbol": symbol, "candidate_id": buy["candidate_id"], "signal_trade_date": date_display(signal_key), "entry_trade_date": buy["entry_trade_date"], "candidate_found": bool(cand), } if not cand: row_out["daily_status"] = "NO_CANDIDATE_ROW" out.append(row_out) continue df = get_daily(symbol) if df is None: row_out["daily_status"] = "DAILY_FILE_MISSING_OR_EMPTY" out.append(row_out) continue idx = index_for_date(df, signal_key) if idx is None: row_out["daily_status"] = "SIGNAL_DATE_NOT_FOUND" out.append(row_out) continue cur = df.iloc[idx] prev5 = df.iloc[max(0, idx - 5) : idx] prev60 = df.iloc[max(0, idx - 60) : idx] lookback30 = df.iloc[max(0, idx - 29) : idx + 1] cur_volume = fnum(cur.get("volume")) prev5_avg = float(prev5["volume"].mean()) if len(prev5) == 5 else None volume_ratio = cur_volume / prev5_avg if cur_volume is not None and prev5_avg and prev5_avg > 0 else None volume_ge2 = volume_ratio is not None and volume_ratio >= 2.0 threshold = board_limit_threshold(symbol) recent_limitup_board_close = False last_limitup_board_close_date = "" recent_limitup_old_proxy = False last_limitup_old_proxy_date = "" for _, drow in lookback30.iterrows(): pre = fnum(drow.get("preClose")) close = fnum(drow.get("close")) high = fnum(drow.get("high")) if pre and pre > 0 and close is not None: pct = (close / pre - 1.0) * 100.0 if pct >= threshold: recent_limitup_board_close = True last_limitup_board_close_date = date_display(str(drow.get("trade_date"))) if pre and pre > 0 and high is not None: proxy_pct = (high / pre - 1.0) * 100.0 if proxy_pct >= 9.5: recent_limitup_old_proxy = True last_limitup_old_proxy_date = date_display(str(drow.get("trade_date"))) valid_prev60 = prev60[prev60["high"].notna()] prev60_high = None prev60_high_volume = None prev60_high_ref_date = "" touch_prev_high = None prev_high_pass = None if len(valid_prev60) >= 20: prev60_high = float(valid_prev60["high"].max()) ref_rows = valid_prev60[valid_prev60["high"] >= prev60_high - 1e-8] if len(ref_rows) > 0: ref = ref_rows.iloc[0] prev60_high_volume = fnum(ref.get("volume")) prev60_high_ref_date = date_display(str(ref.get("trade_date"))) cur_high = fnum(cur.get("high")) touch_prev_high = cur_high is not None and prev60_high is not None and cur_high >= prev60_high * 0.995 if touch_prev_high: prev_high_pass = cur_volume is not None and prev60_high_volume is not None and cur_volume > prev60_high_volume else: prev_high_pass = True else: touch_prev_high = False prev_high_pass = True old_volume_ratio = fnum(cand.get("volume_ratio")) old_prev5 = fnum(cand.get("prev5_avg_volume")) old_recent = bval(cand.get("recent_limitup_30_flag")) old_prev_high_pass = bval(cand.get("prev_high_volume_pass_flag")) old_strict = bval(cand.get("strict_candidate_flag")) strict_recalc_old_field = prev_high_pass row_out.update( { "daily_status": "OK", "daily_file_path": str(daily_path(symbol)), "old_open_price": cand.get("open_price", ""), "recalc_open": cur.get("open", ""), "open_match": near_equal(fnum(cand.get("open_price")), fnum(cur.get("open"))), "old_high_price": cand.get("high_price", ""), "recalc_high": cur.get("high", ""), "high_match": near_equal(fnum(cand.get("high_price")), fnum(cur.get("high"))), "old_low_price": cand.get("low_price", ""), "recalc_low": cur.get("low", ""), "low_match": near_equal(fnum(cand.get("low_price")), fnum(cur.get("low"))), "old_close_price": cand.get("close_price", ""), "recalc_close": cur.get("close", ""), "close_match": near_equal(fnum(cand.get("close_price")), fnum(cur.get("close"))), "old_volume": cand.get("volume", ""), "recalc_volume": cur.get("volume", ""), "volume_match": near_equal(fnum(cand.get("volume")), fnum(cur.get("volume")), rel=1e-8, abs_tol=0.1), "old_prev5_avg_volume": cand.get("prev5_avg_volume", ""), "recalc_prev5_avg_volume": prev5_avg if prev5_avg is not None else "", "prev5_count": len(prev5), "old_volume_ratio": old_volume_ratio if old_volume_ratio is not None else "", "recalc_volume_ratio": volume_ratio if volume_ratio is not None else "", "volume_ratio_match": near_equal(old_volume_ratio, volume_ratio, rel=1e-6, abs_tol=1e-4), "recalc_volume_ratio_ge2": volume_ge2, "limitup_threshold_pct_proxy": threshold, "old_recent_limitup_30_flag": old_recent, "recalc_recent_limitup_30_flag_old_proxy_high_ge_9p5": recent_limitup_old_proxy, "recalc_last_limitup_date_old_proxy": last_limitup_old_proxy_date, "recent_limitup_old_proxy_match": old_recent == recent_limitup_old_proxy if old_recent is not None else "", "recalc_recent_limitup_30_flag_board_close_proxy": recent_limitup_board_close, "recalc_last_limitup_date_board_close_proxy": last_limitup_board_close_date, "old_prev60_high": cand.get("prev60_high", ""), "recalc_prev60_valid_high_count": len(valid_prev60), "recalc_prev60_high": prev60_high if prev60_high is not None else "", "old_prev60_high_volume": cand.get("prev60_high_volume", ""), "recalc_prev60_high_volume": prev60_high_volume if prev60_high_volume is not None else "", "old_prev60_high_ref_date": cand.get("prev60_high_ref_date", ""), "recalc_prev60_high_ref_date": prev60_high_ref_date, "old_touch_prev_high_flag": cand.get("touch_prev_high_flag", ""), "recalc_touch_prev_high_flag": touch_prev_high if touch_prev_high is not None else "", "old_prev_high_volume_pass_flag": old_prev_high_pass, "recalc_prev_high_volume_pass_flag": prev_high_pass if prev_high_pass is not None else "", "prev_high_pass_match": old_prev_high_pass == prev_high_pass if old_prev_high_pass is not None and prev_high_pass is not None else "", "old_strict_candidate_flag": old_strict, "strict_candidate_field_recalc_prev_high_pass_only": strict_recalc_old_field, "strict_candidate_field_match": old_strict == strict_recalc_old_field if old_strict is not None else "", "candidate_rank": cand.get("candidate_rank", ""), "market_gate_open_flag": cand.get("market_gate_open_flag", ""), "old_up_count": cand.get("up_count", ""), } ) out.append(row_out) return out def recalc_market_breadth(source_buy_rows: list[dict[str, Any]], selected: pd.DataFrame) -> list[dict[str, Any]]: selected_by_id = selected.set_index("candidate_id", drop=False).to_dict("index") old_by_date: dict[str, dict[str, Any]] = {} needed_dates: set[str] = set() for buy in source_buy_rows: cand = selected_by_id.get(buy["candidate_id"]) if not cand: continue key = date_key(cand.get("signal_trade_date", "")) if not key: continue needed_dates.add(key) old_by_date.setdefault( key, { "old_stock_count": cand.get("stock_count", ""), "old_up_count": cand.get("up_count", ""), "old_flat_count": cand.get("flat_count", ""), "old_down_count": cand.get("down_count", ""), "old_market_gate_open_flag": cand.get("market_gate_open_flag", ""), }, ) counts: dict[str, Counter[str]] = {key: Counter() for key in needed_dates} for path in DAILY_DIR.glob("*.csv"): try: df = pd.read_csv(path, dtype={"trade_date": str}, usecols=["trade_date", "close", "preClose"]) except Exception: continue df["trade_date"] = df["trade_date"].astype(str).str[:8] df = df[df["trade_date"].isin(needed_dates)] if df.empty: continue df["close"] = pd.to_numeric(df["close"], errors="coerce") df["preClose"] = pd.to_numeric(df["preClose"], errors="coerce") for _, row in df.iterrows(): key = str(row["trade_date"]) close = fnum(row.get("close")) pre = fnum(row.get("preClose")) if close is None or pre is None or pre <= 0: continue counts[key]["stock_count"] += 1 if close > pre: counts[key]["up_count"] += 1 elif close < pre: counts[key]["down_count"] += 1 else: counts[key]["flat_count"] += 1 rows: list[dict[str, Any]] = [] for key in sorted(needed_dates): c = counts[key] old = old_by_date.get(key, {}) old_up = fnum(old.get("old_up_count")) old_gate = bval(old.get("old_market_gate_open_flag")) recalc_gate = c["up_count"] >= 3000 rows.append( { "signal_trade_date": date_display(key), **old, "recalc_stock_count": c["stock_count"], "recalc_up_count": c["up_count"], "recalc_flat_count": c["flat_count"], "recalc_down_count": c["down_count"], "recalc_gate_open_up_count_ge3000": recalc_gate, "old_up_count_diff": c["up_count"] - old_up if old_up is not None else "", "gate_match_old": old_gate == recalc_gate if old_gate is not None else "", } ) return rows def build_minute_events( strict_orders: pd.DataFrame, sell_signals: pd.DataFrame, rolling_signals: pd.DataFrame, ) -> list[dict[str, Any]]: events: list[dict[str, Any]] = [] for row in strict_orders.to_dict("records"): events.append( { "event_source": "STRICT_ORDER", "event_id": row.get("order_id", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "event_trade_date": row.get("trade_date", ""), "date_key": date_key(row.get("trade_date", "")), "event_time": row.get("trade_time", ""), "action_or_signal": row.get("action", ""), "signal_type": row.get("exit_signal_type", ""), } ) for row in sell_signals.to_dict("records"): events.append( { "event_source": "SELL_SIGNAL", "event_id": row.get("signal_id", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "event_trade_date": row.get("observation_trade_date", ""), "date_key": date_key(row.get("observation_trade_date", "")), "event_time": row.get("candidate_time", ""), "action_or_signal": row.get("human_decision_action", ""), "signal_type": row.get("signal_type", ""), } ) for row in rolling_signals.to_dict("records"): events.append( { "event_source": "ROLLING_SIGNAL", "event_id": row.get("rolling_signal_id", ""), "case_id": row.get("case_id", ""), "symbol": row.get("symbol", ""), "event_trade_date": row.get("rolling_trade_date", ""), "date_key": date_key(row.get("rolling_trade_date", "")), "event_time": row.get("rolling_time", ""), "action_or_signal": row.get("human_decision_action", ""), "signal_type": row.get("signal_type", ""), } ) return events def scan_minute_coverage(events: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], dict[tuple[str, str], dict[str, Any]]]: grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) for ev in events: if ev.get("symbol") and ev.get("date_key"): grouped[ev["symbol"]].append(ev) minute_date_stats: dict[tuple[str, str], dict[str, Any]] = {} coverage_rows: list[dict[str, Any]] = [] for symbol, evs in grouped.items(): path = minute_path(symbol) file_status = "OK" if not path.exists(): file_status = "MISSING" elif path.stat().st_size == 0: file_status = "EMPTY" needed_dates = {ev["date_key"] for ev in evs if ev.get("date_key")} needed_times_by_date: dict[str, set[str]] = defaultdict(set) for ev in evs: if ev.get("event_time"): needed_times_by_date[ev["date_key"]].add(ev["event_time"]) stats: dict[str, dict[str, Any]] = { key: { "row_count": 0, "first_time": "", "last_time": "", "times": set(), "exact_rows": {}, "high_until_1040": None, "close_at_1040": None, } for key in needed_dates } if file_status == "OK": with path.open("r", newline="", encoding="utf-8-sig") as f: reader = csv.DictReader(f) for row in reader: timetag = row.get("timetag", "") if len(timetag) < 17: continue dkey = timetag[:8] if dkey not in needed_dates: continue t = timetag[9:17] st = stats[dkey] st["row_count"] += 1 st["times"].add(t) if not st["first_time"] or t < st["first_time"]: st["first_time"] = t if not st["last_time"] or t > st["last_time"]: st["last_time"] = t if t in needed_times_by_date[dkey]: st["exact_rows"][t] = row if t <= "10:40:00": high = fnum(row.get("high")) if high is not None: st["high_until_1040"] = high if st["high_until_1040"] is None else max(st["high_until_1040"], high) if t == "10:40:00": st["close_at_1040"] = fnum(row.get("close")) for dkey in needed_dates: st = stats[dkey] minute_date_stats[(symbol, dkey)] = { "minute_file_status": file_status, "minute_file_path": str(path), "date_row_count": st["row_count"], "first_time": st["first_time"], "last_time": st["last_time"], "high_until_1040": st["high_until_1040"], "close_at_1040": st["close_at_1040"], "times": st["times"], "exact_rows": st["exact_rows"], } for ev in evs: dkey = ev["date_key"] etime = ev.get("event_time", "") st = stats.get(dkey, {}) exact = st.get("exact_rows", {}).get(etime) if etime else None coverage_rows.append( { **{k: v for k, v in ev.items() if k != "date_key"}, "minute_file_status": file_status, "minute_file_path": str(path), "minute_date_row_count": st.get("row_count", 0), "minute_date_found": st.get("row_count", 0) > 0, "event_time_provided": bool(etime), "exact_time_found": bool(exact), "minute_first_time": st.get("first_time", ""), "minute_last_time": st.get("last_time", ""), "event_open": exact.get("open", "") if exact else "", "event_high": exact.get("high", "") if exact else "", "event_low": exact.get("low", "") if exact else "", "event_close": exact.get("close", "") if exact else "", "event_volume": exact.get("volumn", "") if exact else "", "high_until_1040": st.get("high_until_1040", ""), "coverage_status": ( "EXACT_TIME_FOUND" if exact else ( "DATE_FOUND_TIME_MISSING" if st.get("row_count", 0) > 0 else f"NO_DATE_ROWS_{file_status}" ) ), } ) return coverage_rows, minute_date_stats def recalc_three_day_high(sell_signals: pd.DataFrame, minute_date_stats: dict[tuple[str, str], dict[str, Any]]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] three = sell_signals[sell_signals["signal_type"] == "SELL_THREE_DAY_HIGH_NOT_RISING"] for row in three.to_dict("records"): symbol = row.get("symbol", "") obs_key = date_key(row.get("observation_trade_date", "")) out: dict[str, Any] = { "signal_id": row.get("signal_id", ""), "case_id": row.get("case_id", ""), "symbol": symbol, "observation_trade_date": date_display(obs_key), "candidate_time": row.get("candidate_time", ""), "human_decision_action": row.get("human_decision_action", ""), "code_evidence_reason_cn": row.get("code_evidence_reason_cn", ""), } df = get_daily(symbol) if df is None: out["daily_status"] = "DAILY_FILE_MISSING_OR_EMPTY" rows.append(out) continue idx = index_for_date(df, obs_key) if idx is None: out["daily_status"] = "OBS_DATE_NOT_FOUND" rows.append(out) continue if idx < 2: out["daily_status"] = "INSUFFICIENT_PREVIOUS_DAYS" rows.append(out) continue d2 = df.iloc[idx - 2] d1 = df.iloc[idx - 1] cur = df.iloc[idx] h2 = fnum(d2.get("high")) h1 = fnum(d1.get("high")) daily_cur_high = fnum(cur.get("high")) st = minute_date_stats.get((symbol, obs_key), {}) morning_high = st.get("high_until_1040") if morning_high == "": morning_high = None decision_safe = morning_high is not None and st.get("minute_file_status") == "OK" and st.get("date_row_count", 0) > 0 current_for_proxy = morning_high if decision_safe else daily_cur_high not_rising_decision_safe = "" not_rising_proxy = "" if h2 is not None and h1 is not None and decision_safe and current_for_proxy is not None: not_rising_decision_safe = not (h2 < h1 < current_for_proxy) if h2 is not None and h1 is not None and current_for_proxy is not None: not_rising_proxy = not (h2 < h1 < current_for_proxy) out.update( { "daily_status": "OK", "d_minus_2_trade_date": date_display(str(d2.get("trade_date"))), "d_minus_2_high_front": h2 if h2 is not None else "", "d_minus_1_trade_date": date_display(str(d1.get("trade_date"))), "d_minus_1_high_front": h1 if h1 is not None else "", "current_daily_high_front_proxy": daily_cur_high if daily_cur_high is not None else "", "minute_file_status": st.get("minute_file_status", ""), "minute_date_row_count": st.get("date_row_count", ""), "current_high_until_1040_front": morning_high if morning_high is not None else "", "decision_safe_current_high_available": decision_safe, "three_highs_not_strictly_rising_decision_safe": not_rising_decision_safe, "three_highs_not_strictly_rising_daily_proxy": not_rising_proxy, "verification_status": ( "PASS_DECISION_SAFE" if not_rising_decision_safe is True else ( "FAIL_DECISION_SAFE" if not_rising_decision_safe is False else ( "PASS_DAILY_PROXY_NOT_DECISION_SAFE" if not_rising_proxy is True else ( "FAIL_DAILY_PROXY_NOT_DECISION_SAFE" if not_rising_proxy is False else "INSUFFICIENT_DATA" ) ) ) ), } ) rows.append(out) return rows def summarize(rows: list[dict[str, Any]], key: str) -> dict[str, int]: return dict(Counter(str(row.get(key, "")) for row in rows)) def build_summary( strict_orders: pd.DataFrame, sell_signals: pd.DataFrame, rolling_signals: pd.DataFrame, full_orders: pd.DataFrame, source_trace: list[dict[str, Any]], rolling_trace: list[dict[str, Any]], daily_recalc: list[dict[str, Any]], market_breadth: list[dict[str, Any]], minute_coverage: list[dict[str, Any]], three_day: list[dict[str, Any]], ) -> dict[str, Any]: full_order_ids = set(full_orders["order_id"].tolist()) source_buys = [r for r in strict_orders.to_dict("records") if r.get("action") == "BUY" and r.get("source_order_id") in full_order_ids] rolling_buys = [r for r in strict_orders.to_dict("records") if r.get("action") == "BUY" and r.get("source_order_id") not in full_order_ids] daily_ok = [r for r in daily_recalc if r.get("daily_status") == "OK"] summary = { "schema_version": "1.0", "run_id": RUN_ID, "generated_at": datetime.now().isoformat(timespec="seconds"), "sources": { "strict_root": str(STRICT_ROOT), "readable_root": str(READABLE_ROOT), "full_root": str(FULL_ROOT), "daily_root": str(DAILY_ROOT), "minute_root": str(MINUTE_ROOT), }, "scope": { "strict_order_rows": int(len(strict_orders)), "source_buy_orders": int(len(source_buys)), "rolling_buy_orders": int(len(rolling_buys)), "sell_signal_rows": int(len(sell_signals)), "rolling_signal_rows": int(len(rolling_signals)), }, "source_buy_trace": { "rows": len(source_trace), "source_order_found": sum(1 for r in source_trace if r.get("source_order_found") is True), "candidate_found": sum(1 for r in source_trace if r.get("candidate_found") is True), "source_evidence_image_exists": sum(1 for r in source_trace if r.get("source_evidence_image_exists") is True), "normal_buy_time_window_ok": sum(1 for r in source_trace if r.get("normal_buy_time_window_ok") is True), "candidate_rank_top5": sum(1 for r in source_trace if r.get("candidate_rank_top5") is True), "old_market_up_count_ge3000": sum(1 for r in source_trace if r.get("old_up_count_ge3000") is True), }, "rolling_buy_trace": { "rows": len(rolling_trace), "rolling_signal_found": sum(1 for r in rolling_trace if r.get("rolling_signal_found") is True), "rolling_window_1040_1440_ok": sum(1 for r in rolling_trace if r.get("rolling_window_1040_1440_ok") is True), "evidence_image_exists": sum(1 for r in rolling_trace if r.get("evidence_image_exists") is True), }, "daily_candidate_recalc": { "rows": len(daily_recalc), "daily_status_counts": summarize(daily_recalc, "daily_status"), "daily_ok_rows": len(daily_ok), "volume_ratio_ge2": sum(1 for r in daily_ok if r.get("recalc_volume_ratio_ge2") is True), "volume_ratio_lt2_or_unknown": sum(1 for r in daily_ok if r.get("recalc_volume_ratio_ge2") is not True), "recent_limitup_old_proxy_high_ge_9p5_true": sum(1 for r in daily_ok if r.get("recalc_recent_limitup_30_flag_old_proxy_high_ge_9p5") is True), "recent_limitup_board_close_proxy_true": sum(1 for r in daily_ok if r.get("recalc_recent_limitup_30_flag_board_close_proxy") is True), "prev_high_volume_pass_true": sum(1 for r in daily_ok if r.get("recalc_prev_high_volume_pass_flag") is True), "strict_candidate_field_recalc_true": sum(1 for r in daily_ok if r.get("strict_candidate_field_recalc_prev_high_pass_only") is True), "price_ohlc_all_match": sum( 1 for r in daily_ok if r.get("open_match") is True and r.get("high_match") is True and r.get("low_match") is True and r.get("close_match") is True ), "volume_ratio_match": sum(1 for r in daily_ok if r.get("volume_ratio_match") is True), "recent_limitup_old_proxy_match": sum(1 for r in daily_ok if r.get("recent_limitup_old_proxy_match") is True), "prev_high_pass_match": sum(1 for r in daily_ok if r.get("prev_high_pass_match") is True), "strict_candidate_field_match": sum(1 for r in daily_ok if r.get("strict_candidate_field_match") is True), }, "market_breadth_recalc": { "dates": len(market_breadth), "gate_open_recalc_dates": sum(1 for r in market_breadth if r.get("recalc_gate_open_up_count_ge3000") is True), "gate_match_old_dates": sum(1 for r in market_breadth if r.get("gate_match_old") is True), "gate_mismatch_dates": sum(1 for r in market_breadth if r.get("gate_match_old") is False), }, "minute_coverage": { "events": len(minute_coverage), "coverage_status_counts": summarize(minute_coverage, "coverage_status"), "minute_file_status_counts": summarize(minute_coverage, "minute_file_status"), "by_event_source": { src: { "events": sum(1 for r in minute_coverage if r.get("event_source") == src), "exact_time_found": sum(1 for r in minute_coverage if r.get("event_source") == src and r.get("exact_time_found") is True), "date_found": sum(1 for r in minute_coverage if r.get("event_source") == src and r.get("minute_date_found") is True), } for src in sorted({r.get("event_source") for r in minute_coverage}) }, }, "three_day_high_recalc": { "rows": len(three_day), "verification_status_counts": summarize(three_day, "verification_status"), "decision_safe_current_high_available": sum(1 for r in three_day if r.get("decision_safe_current_high_available") is True), }, "boundaries": [ "Daily candidate fields can be recalculated with the supplied front-adjusted daily data.", "The historical candidate generator used high/previous-close >= 9.5% as the recent limit-up memory proxy for all boards; board-specific close-limit recalculation is reported separately and is not the frozen V0/V1 candidate rule.", "Minute-level verification is limited by missing or empty minute files, especially many SZ 000/300 symbols and some SH 688 files.", "SELL_OPEN_VOLUME_STALL has zero rows in the V1 signal ledger; this audit confirms ledger absence but does not prove no historical candidates existed without a dedicated scan.", "Market-risk intraday breadth is not closed by single-symbol minute files; it still needs a whole-market minute breadth dataset or scan.", "Some order prices and old minute chart prices are not directly comparable with the supplied front-adjusted daily data after corporate-action adjustment; rule-direction checks and price-level checks are separated.", ], } return summary def write_markdown(summary: dict[str, Any]) -> None: lines: list[str] = [] lines.append(f"# {RUN_ID}") lines.append("") lines.append("## Scope") for k, v in summary["scope"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Source Buy Trace") for k, v in summary["source_buy_trace"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Rolling Buy Trace") for k, v in summary["rolling_buy_trace"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Daily Candidate Recalc") for k, v in summary["daily_candidate_recalc"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Market Breadth Recalc") for k, v in summary["market_breadth_recalc"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Minute Coverage") for k, v in summary["minute_coverage"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Three-Day High Recalc") for k, v in summary["three_day_high_recalc"].items(): lines.append(f"- {k}: {v}") lines.append("") lines.append("## Boundaries") for item in summary["boundaries"]: lines.append(f"- {item}") lines.append("") lines.append("## Artifacts") for name in [ "source_buy_trace_audit.csv", "rolling_buy_trace_audit.csv", "daily_candidate_rule_recalc.csv", "market_breadth_recalc.csv", "minute_event_coverage_audit.csv", "sell_three_day_high_recalc.csv", "frontdata_rule_audit_summary.json", "manifest.csv", "self_check.json", ]: lines.append(f"- `{name}`") (PACKAGE_ROOT / "README.md").write_text("\n".join(lines) + "\n", encoding="utf-8") def write_manifest() -> None: rows = [] for path in sorted(PACKAGE_ROOT.rglob("*")): if path.is_file(): rows.append( { "path": str(path.relative_to(PACKAGE_ROOT)).replace("\\", "/"), "bytes": path.stat().st_size, } ) write_csv(PACKAGE_ROOT / "manifest.csv", rows, ["path", "bytes"]) def main() -> None: strict_orders = read_csv(STRICT_ROOT / "strict_order_ledger.csv") sell_signals = read_csv(STRICT_ROOT / "strict_sell_signal_ledger.csv") rolling_signals = read_csv(STRICT_ROOT / "rolling_low_buy_signal_ledger.csv") full_orders = read_csv(FULL_ROOT / "order_ledger.csv") selected = read_csv(FULL_ROOT / "selected_candidate_ledger.csv") source_trace = build_source_buy_trace(strict_orders, full_orders, selected) rolling_trace = build_rolling_buy_trace(strict_orders, full_orders, rolling_signals) daily_recalc = recalc_daily_candidate_rules(source_trace, selected) market_breadth = recalc_market_breadth(source_trace, selected) minute_events = build_minute_events(strict_orders, sell_signals, rolling_signals) minute_coverage, minute_date_stats = scan_minute_coverage(minute_events) three_day = recalc_three_day_high(sell_signals, minute_date_stats) write_csv(PACKAGE_ROOT / "source_buy_trace_audit.csv", source_trace) write_csv(PACKAGE_ROOT / "rolling_buy_trace_audit.csv", rolling_trace) write_csv(PACKAGE_ROOT / "daily_candidate_rule_recalc.csv", daily_recalc) write_csv(PACKAGE_ROOT / "market_breadth_recalc.csv", market_breadth) write_csv(PACKAGE_ROOT / "minute_event_coverage_audit.csv", minute_coverage) write_csv(PACKAGE_ROOT / "sell_three_day_high_recalc.csv", three_day) summary = build_summary( strict_orders=strict_orders, sell_signals=sell_signals, rolling_signals=rolling_signals, full_orders=full_orders, source_trace=source_trace, rolling_trace=rolling_trace, daily_recalc=daily_recalc, market_breadth=market_breadth, minute_coverage=minute_coverage, three_day=three_day, ) (PACKAGE_ROOT / "frontdata_rule_audit_summary.json").write_text( json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8", ) write_markdown(summary) self_check = { "run_id": RUN_ID, "status": "PASS", "items": { "source_buy_trace_rows_match_source_buy_orders": summary["source_buy_trace"]["rows"] == summary["scope"]["source_buy_orders"], "rolling_buy_trace_rows_match_rolling_buy_orders": summary["rolling_buy_trace"]["rows"] == summary["scope"]["rolling_buy_orders"], "daily_recalc_rows_match_source_buy_orders": summary["daily_candidate_recalc"]["rows"] == summary["scope"]["source_buy_orders"], "market_breadth_dates_nonzero": summary["market_breadth_recalc"]["dates"] > 0, "minute_coverage_events_nonzero": summary["minute_coverage"]["events"] > 0, "three_day_rows_match_signal_count": summary["three_day_high_recalc"]["rows"] == int((sell_signals["signal_type"] == "SELL_THREE_DAY_HIGH_NOT_RISING").sum()), }, } if not all(self_check["items"].values()): self_check["status"] = "FAIL" (PACKAGE_ROOT / "self_check.json").write_text(json.dumps(self_check, ensure_ascii=False, indent=2), encoding="utf-8") write_manifest() print(json.dumps(summary, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()