from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
import os
|
import re
|
from datetime import datetime
|
from pathlib import Path
|
|
import pandas as pd
|
import pymysql
|
|
|
RUN_ID = "RUN-ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-CHECK-20260614-001"
|
TASK_ID = "ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614"
|
DESIGN_AUDIT_ID = "AUDIT-ANA-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614-DESIGN-001"
|
ISSUE_ID = "ANA-ISSUE-WUJI-V1-ORIGINAL-BUY-LIMITUP-SCOPE-20260614-001"
|
|
ROOT = Path(__file__).resolve().parents[1]
|
PROJECT_ROOT = ROOT.parents[2]
|
SOURCE_V1 = PROJECT_ROOT / "ana-data/result/RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001"
|
SOURCE_FULL = PROJECT_ROOT / "ana-data/result/RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
|
LOCAL_DB_INDEX = Path(r"D:\strategy_project\s-system-doc\observer\天下模型沉淀\数据库索引数据.md")
|
|
STRICT_POLICY = {
|
"window_trading_days": 30,
|
"exclude_signal_day": True,
|
"price_source_table": "a_share_daily_price",
|
"limit_hit_field": "high_price",
|
"previous_close_field": "prev_close",
|
"limit_rate_policy": {
|
"BJ": 0.30,
|
"STAR_688": 0.20,
|
"CHINEXT_300_301": 0.20,
|
"MAINBOARD_DEFAULT": 0.10,
|
},
|
"tolerance_pct_points": 0.05,
|
"limit_hit_formula": "((high_price / prev_close) - 1) * 100 >= limit_rate_pct - tolerance_pct_points",
|
"source_buy_rule": "strict_position_lot_ledger.lot_source_type == SOURCE_BUY and tranche_index == 1",
|
"rolling_buy_rule": "strict_position_lot_ledger.lot_source_type == ROLLING_LOW_BUY and tranche_index > 1",
|
}
|
|
|
def now_iso() -> str:
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
def read_password() -> str:
|
env = os.environ.get("TIANXIA_MYSQL_PASSWORD") or os.environ.get("MYSQL_PWD")
|
if env:
|
return env
|
text = LOCAL_DB_INDEX.read_text(encoding="utf-8")
|
match = re.search(r"^\s*-\s*密码：`([^`]+)`", text, re.MULTILINE)
|
if not match:
|
raise RuntimeError("Unable to read local MySQL credential from approved local index.")
|
return match.group(1)
|
|
|
def get_conn():
|
return pymysql.connect(
|
host="127.0.0.1",
|
port=3306,
|
user="root",
|
password=read_password(),
|
database="tianxia",
|
charset="utf8mb4",
|
connect_timeout=5,
|
read_timeout=180,
|
)
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def rel(path: Path) -> str:
|
return path.relative_to(ROOT).as_posix()
|
|
|
def write_csv(df: pd.DataFrame, name: str) -> Path:
|
path = ROOT / name
|
path.parent.mkdir(parents=True, exist_ok=True)
|
df.to_csv(path, index=False, encoding="utf-8-sig")
|
return path
|
|
|
def write_json(data: dict, name: str) -> Path:
|
path = ROOT / name
|
path.parent.mkdir(parents=True, exist_ok=True)
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
return path
|
|
|
def read_csv(path: Path) -> pd.DataFrame:
|
return pd.read_csv(path, encoding="utf-8-sig")
|
|
|
def board_policy(symbol: str) -> tuple[str, float]:
|
s = str(symbol).upper()
|
code = s.split(".")[0]
|
suffix = s.split(".")[-1] if "." in s else ""
|
if suffix == "BJ" or code.startswith(("8", "4", "9")):
|
return "BJ", 0.30
|
if suffix == "SH" and code.startswith("688"):
|
return "STAR_688", 0.20
|
if suffix == "SZ" and code.startswith(("300", "301")):
|
return "CHINEXT_300_301", 0.20
|
return "MAINBOARD_DEFAULT", 0.10
|
|
|
def clean_str(value: object) -> str:
|
text = "" if pd.isna(value) else str(value).strip()
|
return "" if text.lower() in {"nan", "nat", "none"} else text
|
|
|
def load_scope() -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
|
lots = read_csv(SOURCE_V1 / "strict_position_lot_ledger.csv")
|
orders = read_csv(SOURCE_V1 / "strict_order_ledger.csv")
|
selected = read_csv(SOURCE_FULL / "full_selected_candidate_ledger.csv")
|
|
lots["tranche_index"] = pd.to_numeric(lots["tranche_index"], errors="coerce")
|
source_lots = lots[(lots["lot_source_type"] == "SOURCE_BUY") & (lots["tranche_index"] == 1)].copy()
|
rolling_lots = lots[(lots["lot_source_type"] == "ROLLING_LOW_BUY") & (lots["tranche_index"] > 1)].copy()
|
|
buy_orders = orders[orders["action"] == "BUY"].copy()
|
source_buy_orders = buy_orders[pd.to_numeric(buy_orders["tranche_index"], errors="coerce") == 1].copy()
|
rolling_buy_orders = buy_orders[pd.to_numeric(buy_orders["tranche_index"], errors="coerce") > 1].copy()
|
|
source = source_lots.merge(
|
source_buy_orders[
|
[
|
"order_id",
|
"source_order_id",
|
"case_id",
|
"candidate_id",
|
"symbol",
|
"trade_date",
|
"trade_time",
|
"price",
|
"position_delta_pct",
|
]
|
],
|
left_on=["source_order_id", "case_id", "symbol"],
|
right_on=["source_order_id", "case_id", "symbol"],
|
how="left",
|
suffixes=("_lot", "_order"),
|
)
|
source = source.merge(
|
selected[
|
[
|
"candidate_id",
|
"case_id",
|
"symbol",
|
"signal_trade_date",
|
"entry_trade_date",
|
"candidate_rank",
|
"candidate_status",
|
"strict_candidate_flag",
|
"recent_limitup_30_flag",
|
"last_limitup_date",
|
"days_since_last_limitup",
|
"run_id",
|
"price_source_table",
|
]
|
],
|
on=["candidate_id", "case_id", "symbol"],
|
how="left",
|
suffixes=("", "_candidate"),
|
)
|
return source, rolling_lots, rolling_buy_orders
|
|
|
def query_daily(symbols: list[str], min_date: str, max_date: str) -> pd.DataFrame:
|
placeholders = ",".join(["%s"] * len(symbols))
|
sql = f"""
|
SELECT trade_date, symbol, open_price, high_price, low_price, close_price, volume, amount
|
FROM a_share_daily_price
|
WHERE trade_date BETWEEN %s AND %s
|
AND symbol IN ({placeholders})
|
ORDER BY symbol, trade_date
|
"""
|
params = [min_date, max_date] + symbols
|
with get_conn() as conn:
|
daily = pd.read_sql(sql, conn, params=params)
|
daily["trade_date"] = pd.to_datetime(daily["trade_date"])
|
for col in ["open_price", "high_price", "low_price", "close_price", "volume", "amount"]:
|
daily[col] = pd.to_numeric(daily[col], errors="coerce")
|
daily = daily.sort_values(["symbol", "trade_date"]).reset_index(drop=True)
|
daily["prev_close"] = daily.groupby("symbol")["close_price"].shift(1)
|
return daily
|
|
|
def evaluate_source_buys(source: pd.DataFrame, daily: pd.DataFrame) -> pd.DataFrame:
|
daily_by_symbol = {symbol: g.reset_index(drop=True) for symbol, g in daily.groupby("symbol")}
|
rows = []
|
tol = STRICT_POLICY["tolerance_pct_points"]
|
for row in source.itertuples(index=False):
|
case_id = clean_str(getattr(row, "case_id", ""))
|
symbol = clean_str(getattr(row, "symbol", ""))
|
candidate_id = clean_str(getattr(row, "candidate_id", ""))
|
signal_date_raw = clean_str(getattr(row, "signal_trade_date", ""))
|
entry_date_raw = clean_str(getattr(row, "entry_trade_date_candidate", "")) or clean_str(getattr(row, "entry_trade_date", ""))
|
board, limit_rate = board_policy(symbol)
|
status = "STRICT_LIMITUP_PASS"
|
reason = ""
|
evidence_date = ""
|
evidence_return_pct = ""
|
evidence_high = ""
|
evidence_prev_close = ""
|
window_start = ""
|
window_end = ""
|
window_rows = 0
|
max_return_pct = ""
|
data_gap_reason = ""
|
if not signal_date_raw:
|
status = "SIGNAL_DATE_MISSING_HELD"
|
reason = "候选账本缺 signal_trade_date，无法复核信号日前 30 个交易日涨停。"
|
data_gap_reason = "signal_trade_date_missing"
|
window = pd.DataFrame()
|
elif symbol not in daily_by_symbol:
|
status = "DAILY_DATA_GAP_HELD"
|
reason = "日线源缺该 symbol，无法复核严格涨停记忆。"
|
data_gap_reason = "symbol_daily_missing"
|
window = pd.DataFrame()
|
else:
|
signal_ts = pd.Timestamp(signal_date_raw)
|
g = daily_by_symbol[symbol]
|
prior = g[g["trade_date"] < signal_ts].tail(30).copy()
|
window = prior
|
window_rows = len(prior)
|
if not prior.empty:
|
window_start = prior["trade_date"].iloc[0].date().isoformat()
|
window_end = prior["trade_date"].iloc[-1].date().isoformat()
|
prior["limit_return_pct"] = (prior["high_price"] / prior["prev_close"] - 1.0) * 100.0
|
prior["limit_rate_pct"] = limit_rate * 100.0
|
prior["strict_hit"] = prior["prev_close"].gt(0) & prior["limit_return_pct"].ge(
|
prior["limit_rate_pct"] - tol
|
)
|
max_return_pct = "" if prior["limit_return_pct"].dropna().empty else round(float(prior["limit_return_pct"].max()), 6)
|
hits = prior[prior["strict_hit"]].copy()
|
else:
|
hits = pd.DataFrame()
|
if window_rows < 30:
|
status = "PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD"
|
reason = f"信号日前可用日线不足 30 个交易日，实际 {window_rows}。"
|
data_gap_reason = "prior_window_incomplete"
|
elif hits.empty:
|
status = "STRICT_LIMITUP_FAIL"
|
reason = "信号日前 30 个交易日内未发现按板块阈值触及涨停。"
|
else:
|
hit = hits.iloc[-1]
|
evidence_date = hit["trade_date"].date().isoformat()
|
evidence_return_pct = round(float(hit["limit_return_pct"]), 6)
|
evidence_high = round(float(hit["high_price"]), 6)
|
evidence_prev_close = round(float(hit["prev_close"]), 6)
|
reason = f"信号日前 30 个交易日内于 {evidence_date} 触及板块涨停阈值。"
|
|
rows.append(
|
{
|
"run_id": RUN_ID,
|
"case_id": case_id,
|
"symbol": symbol,
|
"strict_lot_id": clean_str(getattr(row, "strict_lot_id", "")),
|
"source_lot_id": clean_str(getattr(row, "source_lot_id", "")),
|
"source_order_id": clean_str(getattr(row, "source_order_id", "")),
|
"v1_order_id": clean_str(getattr(row, "order_id", "")),
|
"candidate_id": candidate_id,
|
"candidate_rank": clean_str(getattr(row, "candidate_rank", "")),
|
"signal_trade_date": signal_date_raw,
|
"entry_trade_date": entry_date_raw,
|
"board_policy": board,
|
"limit_rate": limit_rate,
|
"limit_rate_pct": limit_rate * 100.0,
|
"tolerance_pct_points": tol,
|
"limit_hit_field": STRICT_POLICY["limit_hit_field"],
|
"previous_close_field": STRICT_POLICY["previous_close_field"],
|
"window_start_trade_date": window_start,
|
"window_end_trade_date": window_end,
|
"window_trading_days": window_rows,
|
"window_excludes_signal_day_flag": True,
|
"strict_limitup_status": status,
|
"strict_limitup_pass_flag": status == "STRICT_LIMITUP_PASS",
|
"strict_limitup_evidence_date": evidence_date,
|
"strict_limitup_evidence_return_pct": evidence_return_pct,
|
"strict_limitup_evidence_high_price": evidence_high,
|
"strict_limitup_evidence_prev_close": evidence_prev_close,
|
"max_prior_30_high_return_pct": max_return_pct,
|
"data_gap_reason": data_gap_reason,
|
"upstream_recent_limitup_30_flag": clean_str(getattr(row, "recent_limitup_30_flag", "")),
|
"upstream_last_limitup_date": clean_str(getattr(row, "last_limitup_date", "")),
|
"upstream_days_since_last_limitup": clean_str(getattr(row, "days_since_last_limitup", "")),
|
"evaluation_reason_cn": reason,
|
}
|
)
|
return pd.DataFrame(rows)
|
|
|
def make_case_summary(detail: pd.DataFrame) -> pd.DataFrame:
|
grouped = detail.groupby("case_id", dropna=False)
|
summary = grouped.agg(
|
source_buy_lots=("strict_lot_id", "count"),
|
strict_pass_lots=("strict_limitup_pass_flag", "sum"),
|
strict_fail_lots=("strict_limitup_status", lambda s: int((s == "STRICT_LIMITUP_FAIL").sum())),
|
held_lots=("strict_limitup_status", lambda s: int(s.astype(str).str.endswith("_HELD").sum())),
|
).reset_index()
|
summary["case_has_any_strict_fail_or_held"] = (summary["strict_fail_lots"] + summary["held_lots"]) > 0
|
summary["case_strict_all_source_buys_pass_flag"] = summary["strict_pass_lots"].eq(summary["source_buy_lots"])
|
summary["case_scope_status"] = summary["case_strict_all_source_buys_pass_flag"].map(
|
{True: "ALL_ORIGINAL_BUY_STRICT_LIMITUP_PASS", False: "HAS_ORIGINAL_BUY_STRICT_LIMITUP_FAIL_OR_HELD"}
|
)
|
return summary
|
|
|
def build_manifest(files: list[Path]) -> pd.DataFrame:
|
rows = []
|
for path in sorted(files, key=lambda p: rel(p)):
|
rows.append(
|
{
|
"path": rel(path),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
return pd.DataFrame(rows)
|
|
|
def main() -> None:
|
ROOT.mkdir(parents=True, exist_ok=True)
|
(ROOT / "tools").mkdir(parents=True, exist_ok=True)
|
source, rolling_lots, rolling_orders = load_scope()
|
symbols = sorted(source["symbol"].dropna().astype(str).unique())
|
min_signal = pd.to_datetime(source["signal_trade_date"]).min()
|
max_signal = pd.to_datetime(source["signal_trade_date"]).max()
|
pull_start = (min_signal - pd.Timedelta(days=80)).date().isoformat()
|
pull_end = max_signal.date().isoformat()
|
daily = query_daily(symbols, pull_start, pull_end)
|
detail = evaluate_source_buys(source, daily)
|
case_summary = make_case_summary(detail)
|
|
board_rows = []
|
for symbol in sorted(detail["symbol"].unique()):
|
board, rate = board_policy(symbol)
|
board_rows.append({"symbol": symbol, "board_policy": board, "limit_rate": rate})
|
board_df = pd.DataFrame(board_rows)
|
|
colleague_expected = {
|
"source_buy_lots": 710,
|
"rolling_low_buy_lots": 25,
|
"strict_pass_lots": 235,
|
"strict_fail_or_nonpass_lots": 475,
|
"cases_with_any_nonpass": 222,
|
}
|
pass_count = int(detail["strict_limitup_pass_flag"].sum())
|
fail_or_held = int(len(detail) - pass_count)
|
cases_nonpass = int((~case_summary["case_strict_all_source_buys_pass_flag"]).sum())
|
mismatch = pd.DataFrame(
|
[
|
{
|
"metric": "source_buy_lots",
|
"colleague_readout": colleague_expected["source_buy_lots"],
|
"this_run_readout": int(len(detail)),
|
"match_flag": int(len(detail)) == colleague_expected["source_buy_lots"],
|
},
|
{
|
"metric": "rolling_low_buy_lots",
|
"colleague_readout": colleague_expected["rolling_low_buy_lots"],
|
"this_run_readout": int(len(rolling_lots)),
|
"match_flag": int(len(rolling_lots)) == colleague_expected["rolling_low_buy_lots"],
|
},
|
{
|
"metric": "strict_pass_lots",
|
"colleague_readout": colleague_expected["strict_pass_lots"],
|
"this_run_readout": pass_count,
|
"match_flag": pass_count == colleague_expected["strict_pass_lots"],
|
},
|
{
|
"metric": "strict_fail_or_nonpass_lots",
|
"colleague_readout": colleague_expected["strict_fail_or_nonpass_lots"],
|
"this_run_readout": fail_or_held,
|
"match_flag": fail_or_held == colleague_expected["strict_fail_or_nonpass_lots"],
|
},
|
{
|
"metric": "cases_with_any_nonpass",
|
"colleague_readout": colleague_expected["cases_with_any_nonpass"],
|
"this_run_readout": cases_nonpass,
|
"match_flag": cases_nonpass == colleague_expected["cases_with_any_nonpass"],
|
},
|
]
|
)
|
|
source_manifest = []
|
for path in [
|
SOURCE_V1 / "strict_position_lot_ledger.csv",
|
SOURCE_V1 / "strict_order_ledger.csv",
|
SOURCE_FULL / "full_selected_candidate_ledger.csv",
|
SOURCE_FULL / "candidate_generation_summary.json",
|
]:
|
source_manifest.append(
|
{
|
"source_run_id": path.parent.name,
|
"path": path.relative_to(PROJECT_ROOT).as_posix(),
|
"size": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
source_manifest_df = pd.DataFrame(source_manifest)
|
|
files: list[Path] = []
|
run_config = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"task_id": TASK_ID,
|
"generated_at": now_iso(),
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"issue_id": ISSUE_ID,
|
"source_runs": {
|
"v1_run": "RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001",
|
"full_run": "RUN-ANA-WUJI-FULL-2023-2026-20260608-001",
|
},
|
"source_buy_identification": {
|
"source_buy_rule": STRICT_POLICY["source_buy_rule"],
|
"rolling_buy_exclusion_rule": STRICT_POLICY["rolling_buy_rule"],
|
"expected_source_buy_lots": 710,
|
"expected_rolling_low_buy_lots": 25,
|
},
|
"signal_date_policy": {
|
"primary_source": "full_selected_candidate_ledger.signal_trade_date joined by candidate_id/case_id/symbol",
|
"entry_date_source": "full_selected_candidate_ledger.entry_trade_date",
|
"missing_or_conflict_status": "SIGNAL_DATE_MISSING_HELD",
|
},
|
"strict_limitup_policy": STRICT_POLICY,
|
"data_sources": {
|
"daily_price_table": "tianxia.a_share_daily_price",
|
"daily_price_fields": ["trade_date", "symbol", "open_price", "high_price", "low_price", "close_price", "volume", "amount"],
|
"source_v1_lot_ledger": (SOURCE_V1 / "strict_position_lot_ledger.csv").relative_to(PROJECT_ROOT).as_posix(),
|
"source_v1_order_ledger": (SOURCE_V1 / "strict_order_ledger.csv").relative_to(PROJECT_ROOT).as_posix(),
|
"source_full_selected_candidate_ledger": (SOURCE_FULL / "full_selected_candidate_ledger.csv").relative_to(PROJECT_ROOT).as_posix(),
|
},
|
"rounding_and_tolerance": {
|
"reason": "A-share price-limit hits are rounded to price ticks; strict threshold uses a 0.05 percentage-point tolerance and exposes exact high/prev_close return.",
|
"tolerance_pct_points": STRICT_POLICY["tolerance_pct_points"],
|
},
|
"boundary_statuses": [
|
"STRICT_LIMITUP_PASS",
|
"STRICT_LIMITUP_FAIL",
|
"SIGNAL_DATE_MISSING_HELD",
|
"DAILY_DATA_GAP_HELD",
|
"PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD",
|
],
|
"citation_boundary": "Execution review is required before this run's readouts can be cited as formal conclusions.",
|
}
|
files.append(write_json(run_config, "run_config.json"))
|
run_config_md = ROOT / "run_config.md"
|
run_config_md.write_text(
|
"\n".join(
|
[
|
f"# {RUN_ID} run_config",
|
"",
|
f"- 设计审计 ID：`{DESIGN_AUDIT_ID}`",
|
f"- 关联问题：`{ISSUE_ID}`",
|
"- 原始新开仓 BUY：`lot_source_type == SOURCE_BUY` 且 `tranche_index == 1`。",
|
"- 滚动低吸 BUY：`lot_source_type == ROLLING_LOW_BUY` 且 `tranche_index > 1`，本轮从新开仓严格涨停复核排除。",
|
"- 信号日来源：`full_selected_candidate_ledger.signal_trade_date`，按 `candidate_id / case_id / symbol` 追溯。",
|
"- 严格涨停窗口：信号日前 30 个交易日，不含信号日。",
|
"- 涨停判定：`a_share_daily_price.high_price / prev_close - 1` 达到板块阈值，容差 0.05 个百分点。",
|
"- 板块阈值：`.BJ` / 8、4、9 开头为 30%；`688*.SH` 为 20%；`300*.SZ` / `301*.SZ` 为 20%；其他默认 10%。",
|
"- 边界：执行审核通过前，不引用本包读数为正式审计结论。",
|
]
|
),
|
encoding="utf-8",
|
)
|
files.append(run_config_md)
|
files.append(write_csv(detail, "original_buy_limitup_scope_check.csv"))
|
files.append(write_csv(case_summary, "case_limitup_scope_summary.csv"))
|
files.append(write_csv(board_df, "symbol_board_policy_check.csv"))
|
files.append(write_csv(mismatch, "mismatch_with_colleague_readout.csv"))
|
files.append(write_csv(source_manifest_df, "source_artifact_manifest.csv"))
|
|
self_checks = [
|
{
|
"check_id": "SOURCE_BUY_LOT_COUNT_IS_710",
|
"status": "PASS" if len(detail) == 710 else "FAIL",
|
"detail": f"source_buy_lots={len(detail)}",
|
},
|
{
|
"check_id": "ROLLING_LOW_BUY_EXCLUDED_COUNT_IS_25",
|
"status": "PASS" if len(rolling_lots) == 25 else "FAIL",
|
"detail": f"rolling_low_buy_lots={len(rolling_lots)}; rolling_buy_orders={len(rolling_orders)}",
|
},
|
{
|
"check_id": "STATUS_TOTAL_EQUALS_710",
|
"status": "PASS" if int(detail["strict_limitup_status"].count()) == 710 else "FAIL",
|
"detail": detail["strict_limitup_status"].value_counts().to_dict(),
|
},
|
{
|
"check_id": "SOURCE_BUY_TRACE_FIELDS_COMPLETE",
|
"status": "PASS"
|
if not detail[["case_id", "symbol", "candidate_id", "signal_trade_date", "entry_trade_date", "strict_lot_id", "v1_order_id"]]
|
.replace("", pd.NA)
|
.isna()
|
.any()
|
.any()
|
else "FAIL",
|
"detail": "case_id/symbol/candidate_id/signal_trade_date/entry_trade_date/strict_lot_id/v1_order_id present for all source BUY rows.",
|
},
|
{
|
"check_id": "WINDOW_EXCLUDES_SIGNAL_DAY",
|
"status": "PASS" if bool(detail["window_excludes_signal_day_flag"].all()) else "FAIL",
|
"detail": "All rows use daily.trade_date < signal_trade_date.",
|
},
|
{
|
"check_id": "STRICT_WINDOW_DAY_COUNT_OR_HELD",
|
"status": "PASS"
|
if bool((detail["window_trading_days"].ge(30) | detail["strict_limitup_status"].eq("PRIOR_30_TRADING_DAY_WINDOW_INCOMPLETE_HELD")).all())
|
else "FAIL",
|
"detail": "Rows with fewer than 30 prior trading days are marked HELD.",
|
},
|
{
|
"check_id": "BOARD_POLICY_ASSIGNED",
|
"status": "PASS" if not detail["board_policy"].isna().any() else "FAIL",
|
"detail": board_df["board_policy"].value_counts().to_dict(),
|
},
|
{
|
"check_id": "MATCHES_COLLEAGUE_READOUT_TABLE_CREATED",
|
"status": "PASS" if not mismatch.empty else "FAIL",
|
"detail": mismatch[["metric", "match_flag"]].to_dict("records"),
|
},
|
{
|
"check_id": "SOURCE_ARTIFACT_MANIFEST_CREATED",
|
"status": "PASS" if len(source_manifest_df) == 4 else "FAIL",
|
"detail": f"source_artifacts={len(source_manifest_df)}",
|
},
|
]
|
self_check_df = pd.DataFrame(self_checks)
|
files.append(write_csv(self_check_df, "self_check_items.csv"))
|
|
summary = {
|
"schema_version": "1.0",
|
"run_id": RUN_ID,
|
"task_id": TASK_ID,
|
"generated_at": now_iso(),
|
"stage": "ORIGINAL_BUY_LIMITUP_SCOPE_CHECK_DONE_READY_FOR_EXEC_REVIEW",
|
"design_audit_id": DESIGN_AUDIT_ID,
|
"issue_id": ISSUE_ID,
|
"source_runs": [
|
"RUN-ANA-WUJI-STRICT-SELL-ROLLING-REPAIR-20260609-001",
|
"RUN-ANA-WUJI-FULL-2023-2026-20260608-001",
|
],
|
"strict_policy": STRICT_POLICY,
|
"readouts": {
|
"source_buy_lots": int(len(detail)),
|
"rolling_low_buy_lots_excluded": int(len(rolling_lots)),
|
"strict_limitup_pass_lots": pass_count,
|
"strict_limitup_fail_or_held_lots": fail_or_held,
|
"strict_limitup_fail_lots": int((detail["strict_limitup_status"] == "STRICT_LIMITUP_FAIL").sum()),
|
"held_lots": int(detail["strict_limitup_status"].astype(str).str.endswith("_HELD").sum()),
|
"cases_total": int(case_summary["case_id"].nunique()),
|
"cases_all_source_buys_pass": int(case_summary["case_strict_all_source_buys_pass_flag"].sum()),
|
"cases_with_any_fail_or_held": cases_nonpass,
|
},
|
"colleague_readout_comparison": mismatch.to_dict("records"),
|
"conclusion_boundary": [
|
"This package only checks original SOURCE_BUY strict prior limit-up memory.",
|
"ROLLING_LOW_BUY lots are excluded from original new-position buy screening.",
|
"This package does not change V1 sell/rolling-ledger correctness or audited V1 return arithmetic.",
|
"Execution review is required before these readouts can be cited as formal audit conclusions.",
|
],
|
}
|
files.append(write_json(summary, "summary.json"))
|
|
summary_md = ROOT / "summary.md"
|
summary_md.write_text(
|
"\n".join(
|
[
|
f"# {RUN_ID} 摘要",
|
"",
|
f"- 生成时间：{summary['generated_at']}",
|
f"- 设计审计 ID：`{DESIGN_AUDIT_ID}`",
|
f"- 关联问题：`{ISSUE_ID}`",
|
f"- 原始新开仓 BUY lot：{len(detail)}",
|
f"- 已排除滚动低吸 BUY lot：{len(rolling_lots)}",
|
f"- 严格涨停通过 lot：{pass_count}",
|
f"- 严格涨停未通过或待审 lot：{fail_or_held}",
|
f"- 至少一只原始 BUY 未通过或待审的 case：{cases_nonpass}",
|
"",
|
"## 口径",
|
"",
|
"严格涨停窗口为信号日前 30 个交易日，不含信号日；价源使用 `a_share_daily_price.high_price / prev_close` 是否达到板块阈值，容差 0.05 个百分点。",
|
"",
|
"## 边界",
|
"",
|
"本包只复核 V1 原始新开仓 BUY 的严格近期涨停记忆，不改变 V1 卖点、滚动低吸、人工裁决、订单、lot、账户账本或已审核收益读数。执行审核通过前不得把本包读数作为正式审计结论引用。",
|
]
|
),
|
encoding="utf-8",
|
)
|
files.append(summary_md)
|
self_check = {
|
"run_id": RUN_ID,
|
"generated_at": summary["generated_at"],
|
"overall_status": "PASS_FOR_EXECUTION_REVIEW_READY"
|
if all(item["status"] == "PASS" for item in self_checks)
|
else "FAIL",
|
"pass_count": sum(1 for item in self_checks if item["status"] == "PASS"),
|
"fail_count": sum(1 for item in self_checks if item["status"] != "PASS"),
|
"items": self_checks,
|
}
|
files.append(write_json(self_check, "self_check.json"))
|
|
# Add this script after generated artifacts exist.
|
files.append(Path(__file__))
|
manifest_df = build_manifest(files)
|
manifest_csv = write_csv(manifest_df, "manifest.csv")
|
write_json({"run_id": RUN_ID, "generated_at": now_iso(), "manifest_self_included": False, "files": manifest_df.to_dict("records")}, "manifest.json")
|
# Manifest files are intentionally not self-listed; self-hashes are not stable after write.
|
manifest_df.to_csv(manifest_csv, index=False, encoding="utf-8-sig")
|
|
print(json.dumps(summary["readouts"], ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
