from __future__ import annotations
|
|
import csv
|
import hashlib
|
import json
|
from collections import Counter, defaultdict
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Any
|
|
import pandas as pd
|
|
|
RUN_ID = "RUN-ANA-WUJI-NOTE-FIT-AUDIT-20260615-001"
|
|
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"
|
FRONT_AUDIT_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-FRONTDATA-RULE-AUDIT-20260613-001"
|
LIFECYCLE_ROOT = RESULT_ROOT / "RUN-ANA-WUJI-V1-STOCK-LIFECYCLE-PACKAGE-20260611-001"
|
|
LEDGER_FIELDS = [
|
"audit_row_id",
|
"run_id",
|
"case_id",
|
"symbol",
|
"subject_type",
|
"subject_id",
|
"rule_id",
|
"rule_category",
|
"audit_status",
|
"automation_level",
|
"evidence_level",
|
"trade_date",
|
"trade_time",
|
"source_file",
|
"source_key",
|
"evidence_summary",
|
"evidence_path",
|
"needs_manual_review",
|
"notes",
|
]
|
|
SUMMARY_FIELDS = [
|
"case_id",
|
"ledger_rows",
|
"pass_count",
|
"weak_pass_count",
|
"proxy_pass_count",
|
"fail_count",
|
"data_gap_count",
|
"manual_required_count",
|
"not_applicable_count",
|
"case_note_fit_status",
|
]
|
|
PILOT_FIELDS = [
|
"pilot_group",
|
"case_id",
|
"account_return_closed_lots",
|
"v1_return_scope",
|
"case_note_fit_status",
|
"fail_count",
|
"manual_required_count",
|
"suggested_review_focus",
|
"case_image_board",
|
"case_stock_lifecycle_board",
|
]
|
|
SELF_CHECK_FIELDS = ["check_id", "status", "details"]
|
|
MANIFEST_FIELDS = ["path", "size_bytes", "sha256"]
|
|
RULE_CATEGORY = {
|
"MKT_OPEN_NEW_UP_3000": "market",
|
"SEL_PRIOR30_LIMITUP_STRICT_BOARD": "selection",
|
"SEL_VOLUME_MULTIPLE": "selection",
|
"SEL_LONG_UPPER_SHADOW_TOP50": "selection",
|
"SEL_BOTTOM_SUPPORT": "selection",
|
"SEL_SIDEWAYS_60D": "selection",
|
"SEL_TOUCH_PREV_HIGH_VOLUME_GT": "selection",
|
"BUY_TIME_WINDOW_NORMAL": "buy",
|
"BUY_OPEN_RUSH_PULLBACK_OPEN_SHRINK_SUPPORT": "buy",
|
"BUY_OPEN_RUSH_PULLBACK_MA20_SHRINK_SUPPORT": "buy",
|
"BUY_NO_CHASE_INTRADAY_SPIKE": "buy",
|
"BUY_ROLLING_LOW_BUY_MA5_EXCEPTION": "buy",
|
"SELL_THREE_DAY_HIGH_NOT_RISING": "sell",
|
"POS_TRANCHE_5_PARTS": "position",
|
"CAP_T1": "lifecycle",
|
}
|
|
|
def now_iso() -> str:
|
return datetime.now().astimezone().isoformat(timespec="seconds")
|
|
|
def read_csv(path: Path) -> pd.DataFrame:
|
if not path.exists():
|
return 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:
|
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 write_json(path: Path, data: dict[str, Any]) -> None:
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
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 fnum(value: Any) -> float | None:
|
text = str(value).strip()
|
if text == "":
|
return None
|
try:
|
return float(text)
|
except ValueError:
|
return None
|
|
|
def status_from_bool(value: bool | None) -> str:
|
if value is True:
|
return "PASS"
|
if value is False:
|
return "FAIL"
|
return "DATA_GAP"
|
|
|
def status_from_proxy_bool(value: bool | None) -> str:
|
if value is True:
|
return "PROXY_PASS"
|
if value is False:
|
return "FAIL"
|
return "DATA_GAP"
|
|
|
def source_rel(path: Path) -> str:
|
try:
|
return path.relative_to(PROJECT_ROOT).as_posix()
|
except ValueError:
|
return str(path)
|
|
|
def key(*parts: Any) -> str:
|
return "|".join(str(p) for p in parts)
|
|
|
class LedgerBuilder:
|
def __init__(self) -> None:
|
self.rows: list[dict[str, Any]] = []
|
|
def add(
|
self,
|
*,
|
case_id: str,
|
symbol: str,
|
subject_type: str,
|
subject_id: str,
|
rule_id: str,
|
audit_status: str,
|
automation_level: str,
|
evidence_level: str,
|
trade_date: str = "",
|
trade_time: str = "",
|
source_file: str = "",
|
source_key: str = "",
|
evidence_summary: str = "",
|
evidence_path: str = "",
|
needs_manual_review: bool = False,
|
notes: str = "",
|
) -> None:
|
self.rows.append(
|
{
|
"audit_row_id": f"NFA-{len(self.rows) + 1:06d}",
|
"run_id": RUN_ID,
|
"case_id": case_id,
|
"symbol": symbol,
|
"subject_type": subject_type,
|
"subject_id": subject_id,
|
"rule_id": rule_id,
|
"rule_category": RULE_CATEGORY.get(rule_id, ""),
|
"audit_status": audit_status,
|
"automation_level": automation_level,
|
"evidence_level": evidence_level,
|
"trade_date": trade_date,
|
"trade_time": trade_time,
|
"source_file": source_file,
|
"source_key": source_key,
|
"evidence_summary": evidence_summary,
|
"evidence_path": evidence_path,
|
"needs_manual_review": str(bool(needs_manual_review)).lower(),
|
"notes": notes,
|
}
|
)
|
|
|
def load_inputs() -> dict[str, pd.DataFrame]:
|
return {
|
"source_buy": read_csv(FRONT_AUDIT_ROOT / "source_buy_trace_audit.csv"),
|
"daily": read_csv(FRONT_AUDIT_ROOT / "daily_candidate_rule_recalc.csv"),
|
"limitup": read_csv(FRONT_AUDIT_ROOT / "limitup_prior30_user_rule_audit.csv"),
|
"rolling": read_csv(FRONT_AUDIT_ROOT / "rolling_buy_trace_audit.csv"),
|
"sell3": read_csv(FRONT_AUDIT_ROOT / "sell_three_day_high_recalc.csv"),
|
"sell3_5m": read_csv(FRONT_AUDIT_ROOT / "three_day_high_5m_proxy_audit.csv"),
|
"orders": read_csv(STRICT_ROOT / "strict_order_ledger.csv"),
|
"lots": read_csv(STRICT_ROOT / "strict_position_lot_ledger.csv"),
|
"strict_case_summary": read_csv(STRICT_ROOT / "strict_case_summary.csv"),
|
"lifecycle": read_csv(LIFECYCLE_ROOT / "stock_lifecycle_index.csv"),
|
}
|
|
|
def index_by(df: pd.DataFrame, *cols: str) -> dict[str, dict[str, Any]]:
|
if df.empty or any(col not in df.columns for col in cols):
|
return {}
|
out: dict[str, dict[str, Any]] = {}
|
for row in df.to_dict("records"):
|
out[key(*(row.get(col, "") for col in cols))] = row
|
return out
|
|
|
def add_source_buy_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None:
|
source_buy = data["source_buy"]
|
daily_by_order = index_by(data["daily"], "strict_order_id")
|
limitup_by_candidate = index_by(data["limitup"], "case_id", "symbol", "candidate_id")
|
source_path = source_rel(FRONT_AUDIT_ROOT / "source_buy_trace_audit.csv")
|
daily_path = source_rel(FRONT_AUDIT_ROOT / "daily_candidate_rule_recalc.csv")
|
limitup_path = source_rel(FRONT_AUDIT_ROOT / "limitup_prior30_user_rule_audit.csv")
|
|
for row in source_buy.to_dict("records"):
|
case_id = row.get("case_id", "")
|
symbol = row.get("symbol", "")
|
order_id = row.get("strict_order_id", "")
|
candidate_id = row.get("candidate_id", "")
|
trade_date = row.get("entry_trade_date", "")
|
trade_time = row.get("entry_time", "")
|
source_key = key(order_id, case_id, symbol)
|
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="MKT_OPEN_NEW_UP_3000",
|
audit_status=status_from_bool(bval(row.get("old_up_count_ge3000", ""))),
|
automation_level="AUTO_STRICT",
|
evidence_level="LEDGER_ONLY",
|
trade_date=trade_date,
|
trade_time=trade_time,
|
source_file=source_path,
|
source_key=source_key,
|
evidence_summary=f"old_up_count={row.get('old_up_count', '')}; market_gate_open_flag={row.get('market_gate_open_flag', '')}",
|
evidence_path=row.get("source_evidence_image_path", ""),
|
)
|
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="BUY_TIME_WINDOW_NORMAL",
|
audit_status=status_from_bool(bval(row.get("normal_buy_time_window_ok", ""))),
|
automation_level="AUTO_STRICT",
|
evidence_level="EXACT",
|
trade_date=trade_date,
|
trade_time=trade_time,
|
source_file=source_path,
|
source_key=source_key,
|
evidence_summary=f"entry_time={trade_time}; normal_buy_time_window_ok={row.get('normal_buy_time_window_ok', '')}",
|
evidence_path=row.get("source_evidence_image_path", ""),
|
)
|
|
rank = fnum(row.get("candidate_rank", ""))
|
rank_status = "DATA_GAP" if rank is None else ("PASS" if rank <= 50 else "FAIL")
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="SEL_LONG_UPPER_SHADOW_TOP50",
|
audit_status=rank_status,
|
automation_level="AUTO_PROXY",
|
evidence_level="LEDGER_ONLY",
|
trade_date=trade_date,
|
trade_time=trade_time,
|
source_file=source_path,
|
source_key=source_key,
|
evidence_summary=f"candidate_rank={row.get('candidate_rank', '')}; note requires top50 long-upper-shadow pool",
|
evidence_path=row.get("source_evidence_image_path", ""),
|
needs_manual_review=True,
|
notes="排名字段可自动核对，但长上影形态和放量质量仍需图证复核。",
|
)
|
|
daily = daily_by_order.get(order_id, {})
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="SEL_VOLUME_MULTIPLE",
|
audit_status=status_from_bool(bval(daily.get("recalc_volume_ratio_ge2", ""))),
|
automation_level="AUTO_STRICT",
|
evidence_level="EXACT",
|
trade_date=daily.get("signal_trade_date", trade_date),
|
trade_time="",
|
source_file=daily_path,
|
source_key=source_key,
|
evidence_summary=(
|
f"recalc_volume_ratio={daily.get('recalc_volume_ratio', '')}; "
|
f"recalc_volume_ratio_ge2={daily.get('recalc_volume_ratio_ge2', '')}; "
|
f"old_volume_ratio={daily.get('old_volume_ratio', row.get('old_volume_ratio', ''))}"
|
),
|
evidence_path=daily.get("daily_file_path", ""),
|
notes="当前严格倍量先按 >=2.0 打标；V1 历史参考口径可能低于 2.0，需分列解释。",
|
)
|
|
touch = bval(daily.get("recalc_touch_prev_high_flag", ""))
|
pass_prev = bval(daily.get("recalc_prev_high_volume_pass_flag", ""))
|
if touch is False:
|
prev_status = "NOT_APPLICABLE"
|
elif touch is True:
|
prev_status = status_from_bool(pass_prev)
|
else:
|
prev_status = "DATA_GAP"
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="SEL_TOUCH_PREV_HIGH_VOLUME_GT",
|
audit_status=prev_status,
|
automation_level="AUTO_STRICT",
|
evidence_level="EXACT",
|
trade_date=daily.get("signal_trade_date", trade_date),
|
source_file=daily_path,
|
source_key=source_key,
|
evidence_summary=(
|
f"touch_prev_high={daily.get('recalc_touch_prev_high_flag', '')}; "
|
f"prev_high_volume_pass={daily.get('recalc_prev_high_volume_pass_flag', '')}; "
|
f"prev_high_ref_date={daily.get('recalc_prev60_high_ref_date', '')}"
|
),
|
evidence_path=daily.get("daily_file_path", ""),
|
)
|
|
limit = limitup_by_candidate.get(key(case_id, symbol, candidate_id), {})
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id="SEL_PRIOR30_LIMITUP_STRICT_BOARD",
|
audit_status=status_from_bool(bval(limit.get("user_strict_close_limitup_prior30_excl_signal", ""))),
|
automation_level="AUTO_STRICT",
|
evidence_level="EXACT",
|
trade_date=limit.get("signal_trade_date", daily.get("signal_trade_date", "")),
|
source_file=limitup_path,
|
source_key=key(case_id, symbol, candidate_id),
|
evidence_summary=(
|
f"board_rule={limit.get('board_rule', '')}; threshold_pct={limit.get('threshold_pct', '')}; "
|
f"strict_close_prior30={limit.get('user_strict_close_limitup_prior30_excl_signal', '')}; "
|
f"last_limitup={limit.get('user_strict_close_last_limitup_date', '')}"
|
),
|
evidence_path=daily.get("daily_file_path", ""),
|
notes="使用用户确认口径：信号日前30个交易日，不含信号日，按板块阈值。",
|
)
|
|
for rule_id, note in [
|
("SEL_BOTTOM_SUPPORT", "底部承接力需要人工看图确认。"),
|
("SEL_SIDEWAYS_60D", "60日横盘或横盘涨停后回调需要日线图形确认。"),
|
("BUY_OPEN_RUSH_PULLBACK_OPEN_SHRINK_SUPPORT", "买入属于回踩开盘价还是其他形态，需要旧买入图确认。"),
|
("BUY_OPEN_RUSH_PULLBACK_MA20_SHRINK_SUPPORT", "买入是否回踩20日均线附近缩量支撑，需要旧买入图确认。"),
|
("BUY_NO_CHASE_INTRADAY_SPIKE", "是否盘中急拉放量追高，需要买入前分时图确认。"),
|
]:
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="SOURCE_BUY",
|
subject_id=order_id,
|
rule_id=rule_id,
|
audit_status="MANUAL_REQUIRED",
|
automation_level="MANUAL_IMAGE",
|
evidence_level="CHART_REQUIRED",
|
trade_date=trade_date,
|
trade_time=trade_time,
|
source_file=source_path,
|
source_key=source_key,
|
evidence_summary=note,
|
evidence_path=row.get("source_evidence_image_path", ""),
|
needs_manual_review=True,
|
notes=row.get("old_decision_reason_cn", ""),
|
)
|
|
|
def add_rolling_buy_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None:
|
rolling = data["rolling"]
|
source_path = source_rel(FRONT_AUDIT_ROOT / "rolling_buy_trace_audit.csv")
|
for row in rolling.to_dict("records"):
|
case_id = row.get("case_id", "")
|
symbol = row.get("symbol", "")
|
order_id = row.get("strict_order_id", "")
|
near_ma5 = fnum(row.get("near_ma5_pct", ""))
|
time_ok = bval(row.get("rolling_window_1040_1440_ok", ""))
|
ma5_status = "DATA_GAP" if near_ma5 is None else ("PASS" if near_ma5 <= 0.012 else "FAIL")
|
combined_status = "PASS" if time_ok is True and ma5_status == "PASS" else ("FAIL" if time_ok is False or ma5_status == "FAIL" else "DATA_GAP")
|
builder.add(
|
case_id=case_id,
|
symbol=symbol,
|
subject_type="ROLLING_BUY",
|
subject_id=order_id,
|
rule_id="BUY_ROLLING_LOW_BUY_MA5_EXCEPTION",
|
audit_status=combined_status,
|
automation_level="AUTO_STRICT+MANUAL_IMAGE",
|
evidence_level="LEDGER_ONLY",
|
trade_date=row.get("rolling_trade_date", ""),
|
trade_time=row.get("rolling_time", ""),
|
source_file=source_path,
|
source_key=key(order_id, row.get("rolling_signal_id", "")),
|
evidence_summary=(
|
f"rolling_window_1040_1440_ok={row.get('rolling_window_1040_1440_ok', '')}; "
|
f"near_ma5_pct={row.get('near_ma5_pct', '')}; "
|
f"volume_ratio_vs_prev20m={row.get('volume_ratio_vs_prev20m', '')}"
|
),
|
evidence_path=row.get("evidence_image_path", ""),
|
needs_manual_review=True,
|
notes="时间和MA5附近度可自动判，趋势基础和止跌承接仍需人工看图。",
|
)
|
|
|
def add_sell_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None:
|
sell3 = data["sell3"]
|
proxy_by_signal = index_by(data["sell3_5m"], "signal_id")
|
source_path = source_rel(FRONT_AUDIT_ROOT / "sell_three_day_high_recalc.csv")
|
proxy_path = source_rel(FRONT_AUDIT_ROOT / "three_day_high_5m_proxy_audit.csv")
|
for row in sell3.to_dict("records"):
|
signal_id = row.get("signal_id", "")
|
proxy = proxy_by_signal.get(signal_id, {})
|
if proxy:
|
status = status_from_proxy_bool(bval(proxy.get("three_highs_not_strictly_rising_5m_proxy", "")))
|
evidence_level = "PROXY_5M"
|
source_file = proxy_path
|
summary = (
|
f"verification_status_5m_proxy={proxy.get('verification_status_5m_proxy', '')}; "
|
f"three_highs_not_strictly_rising_5m_proxy={proxy.get('three_highs_not_strictly_rising_5m_proxy', '')}; "
|
f"d-2={proxy.get('d_minus_2_high_front', '')}; d-1={proxy.get('d_minus_1_high_front', '')}; "
|
f"current_5m_scaled={proxy.get('current_high_until_1040_5m_daily_scaled_front', '')}"
|
)
|
else:
|
safe = bval(row.get("three_highs_not_strictly_rising_decision_safe", ""))
|
daily_proxy = bval(row.get("three_highs_not_strictly_rising_daily_proxy", ""))
|
if safe is not None:
|
status = status_from_bool(safe)
|
evidence_level = "EXACT"
|
else:
|
status = status_from_proxy_bool(daily_proxy)
|
evidence_level = "DAILY_PROXY"
|
source_file = source_path
|
summary = (
|
f"verification_status={row.get('verification_status', '')}; "
|
f"decision_safe={row.get('three_highs_not_strictly_rising_decision_safe', '')}; "
|
f"daily_proxy={row.get('three_highs_not_strictly_rising_daily_proxy', '')}"
|
)
|
builder.add(
|
case_id=row.get("case_id", ""),
|
symbol=row.get("symbol", ""),
|
subject_type="SELL_SIGNAL",
|
subject_id=signal_id,
|
rule_id="SELL_THREE_DAY_HIGH_NOT_RISING",
|
audit_status=status,
|
automation_level="AUTO_STRICT_OR_PROXY",
|
evidence_level=evidence_level,
|
trade_date=row.get("observation_trade_date", ""),
|
trade_time=row.get("candidate_time", ""),
|
source_file=source_file,
|
source_key=signal_id,
|
evidence_summary=summary,
|
evidence_path=proxy.get("public_5m_cache_path", ""),
|
notes=row.get("code_evidence_reason_cn", ""),
|
)
|
|
|
def add_order_and_lot_rules(builder: LedgerBuilder, data: dict[str, pd.DataFrame]) -> None:
|
orders = data["orders"]
|
lots = data["lots"]
|
order_path = source_rel(STRICT_ROOT / "strict_order_ledger.csv")
|
lot_path = source_rel(STRICT_ROOT / "strict_position_lot_ledger.csv")
|
|
for row in orders.to_dict("records"):
|
if row.get("action") != "BUY":
|
continue
|
planned = fnum(row.get("planned_tranche_count", ""))
|
status = "DATA_GAP" if planned is None else ("PASS" if int(planned) == 5 else "FAIL")
|
builder.add(
|
case_id=row.get("case_id", ""),
|
symbol=row.get("symbol", ""),
|
subject_type="ORDER",
|
subject_id=row.get("order_id", ""),
|
rule_id="POS_TRANCHE_5_PARTS",
|
audit_status=status,
|
automation_level="AUTO_STRICT",
|
evidence_level="LEDGER_ONLY",
|
trade_date=row.get("trade_date", ""),
|
trade_time=row.get("trade_time", ""),
|
source_file=order_path,
|
source_key=row.get("order_id", ""),
|
evidence_summary=(
|
f"tranche_index={row.get('tranche_index', '')}; "
|
f"planned_tranche_count={row.get('planned_tranche_count', '')}; "
|
f"position_delta_pct={row.get('position_delta_pct', '')}"
|
),
|
evidence_path=row.get("evidence_image_path", ""),
|
)
|
|
for row in lots.to_dict("records"):
|
entry = row.get("entry_trade_date", "")
|
sellable = row.get("sellable_from_trade_date", "")
|
exit_date = row.get("exit_trade_date", "")
|
if not entry or not sellable:
|
status = "DATA_GAP"
|
elif exit_date and exit_date < sellable:
|
status = "FAIL"
|
else:
|
status = "PASS"
|
builder.add(
|
case_id=row.get("case_id", ""),
|
symbol=row.get("symbol", ""),
|
subject_type="LOT",
|
subject_id=row.get("strict_lot_id", ""),
|
rule_id="CAP_T1",
|
audit_status=status,
|
automation_level="AUTO_STRICT",
|
evidence_level="LEDGER_ONLY",
|
trade_date=entry,
|
trade_time=row.get("entry_time", ""),
|
source_file=lot_path,
|
source_key=row.get("strict_lot_id", ""),
|
evidence_summary=f"entry={entry}; sellable_from={sellable}; exit={exit_date}; lot_status={row.get('lot_status', '')}",
|
)
|
|
|
def build_summary(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
by_case: dict[str, Counter[str]] = defaultdict(Counter)
|
for row in rows:
|
by_case[row["case_id"]][row["audit_status"]] += 1
|
by_case[row["case_id"]]["_rows"] += 1
|
|
out: list[dict[str, Any]] = []
|
for case_id in sorted(by_case):
|
c = by_case[case_id]
|
if c["FAIL"]:
|
status = "NOTE_MISMATCH_REVIEW"
|
elif c["DATA_GAP"]:
|
status = "DATA_GAP_HELD"
|
elif c["MANUAL_REQUIRED"]:
|
status = "MANUAL_REVIEW_PENDING"
|
elif c["PROXY_PASS"]:
|
status = "NOTE_FIT_PROXY"
|
else:
|
status = "NOTE_FIT_AUTO_PASS"
|
out.append(
|
{
|
"case_id": case_id,
|
"ledger_rows": c["_rows"],
|
"pass_count": c["PASS"],
|
"weak_pass_count": c["WEAK_PASS"],
|
"proxy_pass_count": c["PROXY_PASS"],
|
"fail_count": c["FAIL"],
|
"data_gap_count": c["DATA_GAP"],
|
"manual_required_count": c["MANUAL_REQUIRED"],
|
"not_applicable_count": c["NOT_APPLICABLE"],
|
"case_note_fit_status": status,
|
}
|
)
|
return out
|
|
|
def sample_evenly(rows: list[dict[str, Any]], n: int) -> list[dict[str, Any]]:
|
if len(rows) <= n:
|
return rows
|
if n <= 1:
|
return [rows[0]]
|
indexes = [round(i * (len(rows) - 1) / (n - 1)) for i in range(n)]
|
out: list[dict[str, Any]] = []
|
seen: set[int] = set()
|
for idx in indexes:
|
if idx not in seen:
|
out.append(rows[idx])
|
seen.add(idx)
|
return out
|
|
|
def build_pilot_queue(case_summary: list[dict[str, Any]], data: dict[str, pd.DataFrame]) -> list[dict[str, Any]]:
|
summary_by_case = {row["case_id"]: row for row in case_summary}
|
strict_case = data["strict_case_summary"]
|
strict_by_case = strict_case.set_index("case_id", drop=False).to_dict("index") if not strict_case.empty else {}
|
|
lifecycle_cases = set(data["lifecycle"]["case_id"].tolist()) if not data["lifecycle"].empty else set()
|
|
merged: list[dict[str, Any]] = []
|
for case_id, note in summary_by_case.items():
|
strict = strict_by_case.get(case_id, {})
|
return_value = fnum(strict.get("account_return_closed_lots", ""))
|
case_image = STRICT_ROOT / "cases" / case_id / "case_image_board.md"
|
lifecycle_board = LIFECYCLE_ROOT / "cases" / case_id / "case_stock_lifecycle_board.md"
|
merged.append(
|
{
|
"case_id": case_id,
|
"account_return_closed_lots": "" if return_value is None else f"{return_value:.8f}",
|
"return_float": return_value,
|
"v1_return_scope": strict.get("v1_return_scope", ""),
|
"case_note_fit_status": note.get("case_note_fit_status", ""),
|
"fail_count": note.get("fail_count", "0"),
|
"manual_required_count": note.get("manual_required_count", "0"),
|
"case_image_board": source_rel(case_image) if case_image.exists() else "",
|
"case_stock_lifecycle_board": source_rel(lifecycle_board) if case_id in lifecycle_cases or lifecycle_board.exists() else "",
|
}
|
)
|
|
primary = [r for r in merged if r["v1_return_scope"] == "V1_PRIMARY_STRICT_CLOSED_CASE"]
|
winners = sorted([r for r in primary if r["return_float"] is not None and r["return_float"] > 0], key=lambda r: r["case_id"])
|
losers = sorted([r for r in primary if r["return_float"] is not None and r["return_float"] <= 0], key=lambda r: r["case_id"])
|
manual_only = sorted(
|
[r for r in merged if r["case_note_fit_status"] == "MANUAL_REVIEW_PENDING"],
|
key=lambda r: r["case_id"],
|
)
|
|
out: list[dict[str, Any]] = []
|
for group, source_rows, focus in [
|
("WINNER_SAMPLE", sample_evenly(winners, 10), "盈利样本：检查自动严格差异是否影响成功样本解释。"),
|
("LOSER_SAMPLE", sample_evenly(losers, 10), "亏损/非正收益样本：检查失败是否来自选股、买点、卖点还是市场环境。"),
|
("MANUAL_ONLY_SAMPLE", sample_evenly(manual_only, 10), "无自动失败但需要人工看图样本：校准形态类规则。"),
|
]:
|
for row in source_rows:
|
out.append(
|
{
|
"pilot_group": group,
|
"case_id": row["case_id"],
|
"account_return_closed_lots": row["account_return_closed_lots"],
|
"v1_return_scope": row["v1_return_scope"],
|
"case_note_fit_status": row["case_note_fit_status"],
|
"fail_count": row["fail_count"],
|
"manual_required_count": row["manual_required_count"],
|
"suggested_review_focus": focus,
|
"case_image_board": row["case_image_board"],
|
"case_stock_lifecycle_board": row["case_stock_lifecycle_board"],
|
}
|
)
|
return out
|
|
|
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 build_manifest() -> list[dict[str, Any]]:
|
rows: list[dict[str, Any]] = []
|
for path in sorted(PACKAGE_ROOT.rglob("*")):
|
if not path.is_file():
|
continue
|
if path.name in {"manifest.csv"}:
|
continue
|
rows.append(
|
{
|
"path": source_rel(path),
|
"size_bytes": path.stat().st_size,
|
"sha256": sha256_file(path),
|
}
|
)
|
return rows
|
|
|
def build_self_check(rows: list[dict[str, Any]], case_summary: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
package_files = [
|
"README.md",
|
"todo_status.md",
|
"note_rule_mapping.csv",
|
"data_source_inventory.csv",
|
"output_schema.md",
|
"case_note_fit_ledger.csv",
|
"case_note_fit_summary.csv",
|
"manual_review_queue.csv",
|
"rule_exception_inventory.csv",
|
"pilot_case_review_queue.csv",
|
"note_fit_audit_summary.md",
|
"summary.json",
|
]
|
checks: list[dict[str, Any]] = []
|
|
missing = [name for name in package_files if not (PACKAGE_ROOT / name).exists()]
|
checks.append(
|
{
|
"check_id": "PACKAGE_FILES_EXIST",
|
"status": "PASS" if not missing else "FAIL",
|
"details": "all required files exist" if not missing else f"missing={missing}",
|
}
|
)
|
|
checks.append(
|
{
|
"check_id": "LEDGER_NON_EMPTY",
|
"status": "PASS" if rows else "FAIL",
|
"details": f"ledger_rows={len(rows)}",
|
}
|
)
|
|
bad_pass = [row for row in rows if row["audit_status"] in {"PASS", "PROXY_PASS"} and not row["source_file"]]
|
checks.append(
|
{
|
"check_id": "PASS_ROWS_HAVE_SOURCE",
|
"status": "PASS" if not bad_pass else "FAIL",
|
"details": f"bad_rows={len(bad_pass)}",
|
}
|
)
|
|
manual_rows = [row for row in rows if row["audit_status"] == "MANUAL_REQUIRED" or row["needs_manual_review"] == "true"]
|
manual_file_rows = read_csv(PACKAGE_ROOT / "manual_review_queue.csv")
|
checks.append(
|
{
|
"check_id": "MANUAL_QUEUE_MATCHES_LEDGER",
|
"status": "PASS" if len(manual_rows) == len(manual_file_rows) else "FAIL",
|
"details": f"ledger_manual_rows={len(manual_rows)}; file_rows={len(manual_file_rows)}",
|
}
|
)
|
|
exception_rows = [row for row in rows if row["audit_status"] == "FAIL"]
|
exception_file_rows = read_csv(PACKAGE_ROOT / "rule_exception_inventory.csv")
|
checks.append(
|
{
|
"check_id": "EXCEPTION_INVENTORY_MATCHES_FAILS",
|
"status": "PASS" if len(exception_rows) == len(exception_file_rows) else "FAIL",
|
"details": f"ledger_fail_rows={len(exception_rows)}; file_rows={len(exception_file_rows)}",
|
}
|
)
|
|
checks.append(
|
{
|
"check_id": "CASE_SUMMARY_NON_EMPTY",
|
"status": "PASS" if case_summary else "FAIL",
|
"details": f"case_count={len(case_summary)}",
|
}
|
)
|
|
return checks
|
|
|
def write_summary_markdown(rows: list[dict[str, Any]], case_summary: list[dict[str, Any]]) -> None:
|
status_counts = Counter(row["audit_status"] for row in rows)
|
rule_counts = Counter(row["rule_id"] for row in rows)
|
case_status_counts = Counter(row["case_note_fit_status"] for row in case_summary)
|
lines = [
|
"# Note Fit Audit Summary",
|
"",
|
f"generated_at: {now_iso()}",
|
f"run_id: `{RUN_ID}`",
|
"",
|
"## Scope",
|
"",
|
f"- ledger_rows: {len(rows)}",
|
f"- cases: {len(case_summary)}",
|
"",
|
"## Audit Status Counts",
|
"",
|
]
|
for status, count in status_counts.most_common():
|
lines.append(f"- {status}: {count}")
|
lines.extend(["", "## Case Status Counts", ""])
|
for status, count in case_status_counts.most_common():
|
lines.append(f"- {status}: {count}")
|
lines.extend(["", "## Rule Row Counts", ""])
|
for rule_id, count in rule_counts.most_common():
|
lines.append(f"- {rule_id}: {count}")
|
lines.extend(
|
[
|
"",
|
"## Boundary",
|
"",
|
"This first-pass ledger only handles rules with existing structured evidence. "
|
"Rows marked `MANUAL_REQUIRED` must be reviewed with charts before any strict note-fit conclusion is made.",
|
"",
|
]
|
)
|
(PACKAGE_ROOT / "note_fit_audit_summary.md").write_text("\n".join(lines), encoding="utf-8")
|
|
|
def main() -> None:
|
data = load_inputs()
|
builder = LedgerBuilder()
|
add_source_buy_rules(builder, data)
|
add_rolling_buy_rules(builder, data)
|
add_sell_rules(builder, data)
|
add_order_and_lot_rules(builder, data)
|
|
rows = builder.rows
|
write_csv(PACKAGE_ROOT / "case_note_fit_ledger.csv", rows, LEDGER_FIELDS)
|
|
case_summary = build_summary(rows)
|
write_csv(PACKAGE_ROOT / "case_note_fit_summary.csv", case_summary, SUMMARY_FIELDS)
|
|
manual_rows = [
|
row for row in rows
|
if row["audit_status"] == "MANUAL_REQUIRED" or row["needs_manual_review"] == "true"
|
]
|
write_csv(PACKAGE_ROOT / "manual_review_queue.csv", manual_rows, LEDGER_FIELDS)
|
|
exception_rows = [row for row in rows if row["audit_status"] == "FAIL"]
|
write_csv(PACKAGE_ROOT / "rule_exception_inventory.csv", exception_rows, LEDGER_FIELDS)
|
|
pilot_rows = build_pilot_queue(case_summary, data)
|
write_csv(PACKAGE_ROOT / "pilot_case_review_queue.csv", pilot_rows, PILOT_FIELDS)
|
|
write_summary_markdown(rows, case_summary)
|
write_json(
|
PACKAGE_ROOT / "summary.json",
|
{
|
"generated_at": now_iso(),
|
"run_id": RUN_ID,
|
"ledger_rows": len(rows),
|
"case_count": len(case_summary),
|
"status_counts": dict(Counter(row["audit_status"] for row in rows)),
|
"case_status_counts": dict(Counter(row["case_note_fit_status"] for row in case_summary)),
|
"source_files": {
|
name: len(df)
|
for name, df in data.items()
|
},
|
},
|
)
|
|
self_check_rows = build_self_check(rows, case_summary)
|
write_csv(PACKAGE_ROOT / "self_check_items.csv", self_check_rows, SELF_CHECK_FIELDS)
|
write_json(
|
PACKAGE_ROOT / "self_check.json",
|
{
|
"generated_at": now_iso(),
|
"run_id": RUN_ID,
|
"status": "PASS" if all(row["status"] == "PASS" for row in self_check_rows) else "FAIL",
|
"items": self_check_rows,
|
},
|
)
|
self_check_lines = ["# Self Check", ""]
|
for row in self_check_rows:
|
self_check_lines.append(f"- {row['check_id']}: {row['status']} - {row['details']}")
|
self_check_lines.append("")
|
(PACKAGE_ROOT / "self_check.md").write_text("\n".join(self_check_lines), encoding="utf-8")
|
|
write_csv(PACKAGE_ROOT / "manifest.csv", build_manifest(), MANIFEST_FIELDS)
|
|
|
if __name__ == "__main__":
|
main()
|
