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()
|
