from __future__ import annotations

import hashlib
import json
import re
from datetime import datetime
from pathlib import Path
from urllib.parse import unquote

import pandas as pd


RUN_ID = "RUN-ANA-WUJI-FULL-2023-2026-20260608-001"
ROOT = Path(__file__).resolve().parents[1]
ALLOWED_UNRESOLVED = {"WINDOW_END_VALUATION_ONLY", "EXIT_DATA_GAP_HELD", "EXIT_REVIEW_HELD"}
EXPECTED_ROLE_COUNTS = {
    "candidate_daily_100d_decision_view": "selected_candidate_rows",
    "entry_1m_buy_decision_view": "buy_orders",
    "exit_daily_signal_review_view": "lots",
    "exit_1m_sell_decision_view": "sell_orders",
}


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 check(name: str, passed: bool, detail: str, rows: list[dict]) -> None:
    rows.append({"check_name": name, "status": "PASS" if passed else "FAIL", "detail": detail})


def read_csv(name: str) -> pd.DataFrame:
    return pd.read_csv(ROOT / name, encoding="utf-8-sig")


def as_bool_false(series: pd.Series) -> bool:
    return not series.astype(str).str.lower().isin(["true", "1", "yes"]).any()


def load_optional_csv(name: str) -> pd.DataFrame:
    path = ROOT / name
    return pd.read_csv(path, encoding="utf-8-sig") if path.exists() else pd.DataFrame()


def local_markdown_links(path: Path) -> list[Path]:
    text = path.read_text(encoding="utf-8")
    links: list[Path] = []
    for raw in re.findall(r"\]\(([^)]+)\)", text):
        if "://" in raw or raw.startswith("#"):
            continue
        target = unquote(raw.split("#", 1)[0]).strip()
        if not target:
            continue
        links.append((path.parent / target).resolve())
    return links


def board_status_matches_summary(case_index: pd.DataFrame, lots: pd.DataFrame) -> tuple[bool, str]:
    board_paths = [
        ROOT / "case_image_board.md",
        ROOT / "case_story_board.md",
        *sorted((ROOT / "cases").glob("*/case_image_board.md")),
        *sorted((ROOT / "cases").glob("*/case_story_board.md")),
    ]
    missing = [p.relative_to(ROOT).as_posix() for p in board_paths if not p.exists()]
    stale_terms = [
        "STRUCTURE_PILOT_EXIT_REVIEW_RESOLVED_SELF_CHECK_DONE",
        "7 个小样本案例",
        "退出复核执行审核已通过",
        "当前仍为结构试点",
    ]
    stale_hits: list[str] = []
    required_missing: list[str] = []

    case_count = len(case_index)
    anchor_count = int((case_index.expand_case_role == "ANCHOR_REUSED_FROM_AUDITED_7_CASE_PILOT").sum())
    new_count = int((case_index.expand_case_role == "NEW_EXPAND_30_CASE").sum())
    boundary_count = int((lots.lot_status != "CLOSED_BY_AI_SELL").sum())

    root_required = [
        RUN_ID,
        "EXPAND_30_EXECUTION_SELF_CHECK_DONE",
        f"{case_count} 个案例日",
        f"{anchor_count} 个已审核锚点",
        f"{new_count} 个新增分层案例日",
        f"{boundary_count} 笔",
        "HELD",
        "待复审",
        "RETURN_STAT_READY=false",
    ]
    case_required = [
        RUN_ID,
        "EXPAND_30_EXECUTION_SELF_CHECK_DONE",
        "扩样执行包",
        "待复审",
        "RETURN_STAT_READY=false",
    ]

    for path in board_paths:
        if not path.exists():
            continue
        rel = path.relative_to(ROOT).as_posix()
        text = path.read_text(encoding="utf-8")
        for term in stale_terms:
            if term in text:
                stale_hits.append(f"{rel}:{term}")
        required = root_required if path.parent == ROOT else case_required
        for term in required:
            if term not in text:
                required_missing.append(f"{rel}:{term}")

    ok = not missing and not stale_hits and not required_missing
    detail = (
        f"boards={len(board_paths)}, missing={len(missing)}, stale_hits={len(stale_hits)}, "
        f"required_missing={len(required_missing)}, cases={case_count}, anchors={anchor_count}, "
        f"new={new_count}, boundary_lots={boundary_count}"
    )
    if missing:
        detail += "; missing=" + "|".join(missing[:5])
    if stale_hits:
        detail += "; stale=" + "|".join(stale_hits[:5])
    if required_missing:
        detail += "; required_missing=" + "|".join(required_missing[:5])
    return ok, detail


def main() -> None:
    checks: list[dict] = []
    case_index = read_csv("case_index.csv")
    selected = read_csv("selected_candidate_ledger.csv")
    expand_selected = read_csv("expand_candidate_selection_ledger.csv")
    selection_log = read_csv("expand_sample_selection_log.csv")
    order_ledger = read_csv("order_ledger.csv")
    lots = read_csv("position_lot_ledger.csv")
    decisions = read_csv("decision_log.csv")
    sell_decisions = read_csv("sell_decision_log.csv")
    exit_resolution = read_csv("exit_resolution_log.csv")
    account = read_csv("daily_account_ledger.csv")
    case_summary = read_csv("case_summary.csv")
    image_manifest = read_csv("image_manifest.csv")

    action_counts = order_ledger.action.value_counts().to_dict()
    closed = lots[lots.lot_status == "CLOSED_BY_AI_SELL"].copy()
    unresolved = lots[lots.lot_status != "CLOSED_BY_AI_SELL"].copy()
    buy_orders = int(action_counts.get("BUY", 0))
    sell_orders = int(action_counts.get("SELL", 0))

    anchor_count = int((case_index.expand_case_role == "ANCHOR_REUSED_FROM_AUDITED_7_CASE_PILOT").sum())
    new_count = int((case_index.expand_case_role == "NEW_EXPAND_30_CASE").sum())
    check(
        "expand_case_scope_frozen",
        len(case_index) == 30 and case_index.entry_trade_date.nunique() == 30 and anchor_count == 7 and new_count == 23,
        f"cases={len(case_index)}, unique_dates={case_index.entry_trade_date.nunique()}, anchors={anchor_count}, new={new_count}",
        checks,
    )
    check(
        "expand_selection_no_silent_substitution",
        not selection_log.substitution_flag.astype(str).str.lower().isin(["true", "1", "yes"]).any(),
        f"selection_rows={len(selection_log)}",
        checks,
    )
    check(
        "selected_candidate_scope",
        len(selected) == 150 and selected.case_id.nunique() == 30 and len(expand_selected) == 150,
        f"selected={len(selected)}, frozen_selected={len(expand_selected)}, cases={selected.case_id.nunique()}",
        checks,
    )

    check(
        "order_counts",
        buy_orders == len(lots) and sell_orders == len(closed) and len(order_ledger) == buy_orders + sell_orders,
        f"orders={action_counts}; lots={len(lots)}; closed_lots={len(closed)}",
        checks,
    )

    stage_counts = decisions.groupby(["decision_stage", "action_status"]).size().to_dict()
    check(
        "decision_log_has_entry_and_exit",
        decisions.decision_stage.isin(["ENTRY_AI_REVIEW", "EXIT_AI_REVIEW"]).all()
        and int((decisions.decision_stage == "ENTRY_AI_REVIEW").sum()) == len(selected),
        str(stage_counts),
        checks,
    )

    closed_market_cases = case_index[case_index.market_gate_status == "NO_TRADE_MARKET_GATE_CLOSED"].case_id.tolist()
    no_trade_decisions = decisions[
        (decisions.case_id.isin(closed_market_cases))
        & (decisions.action_status == "NO_TRADE_MARKET_GATE_CLOSED")
    ]
    closed_market_lots = lots[lots.case_id.isin(closed_market_cases)]
    check(
        "market_gate_closed_cases_preserved_no_trade",
        len(no_trade_decisions) == len(closed_market_cases) * 5 and closed_market_lots.empty,
        f"closed_market_cases={len(closed_market_cases)}, no_trade_decisions={len(no_trade_decisions)}, lots={len(closed_market_lots)}",
        checks,
    )

    check(
        "lot_status_counts",
        len(closed) + len(unresolved) == len(lots) and len(lots) == buy_orders,
        f"closed={len(closed)}, unresolved={len(unresolved)}, lots={len(lots)}",
        checks,
    )
    check(
        "unresolved_status_explicit",
        set(unresolved.lot_status).issubset(ALLOWED_UNRESOLVED),
        ",".join(sorted(set(unresolved.lot_status))),
        checks,
    )
    check(
        "exit_resolution_all_lots_explicit",
        len(exit_resolution) == len(lots) and not exit_resolution.final_action_status.isna().any(),
        f"resolution_rows={len(exit_resolution)}; statuses={exit_resolution.final_action_status.value_counts().to_dict()}",
        checks,
    )

    sell_orders_df = order_ledger[order_ledger.action == "SELL"].copy()
    merged = sell_orders_df.merge(lots, left_on="source_lot_id", right_on="trade_lot_id", suffixes=("_order", "_lot"))
    if len(merged):
        t1_ok = (
            (pd.to_datetime(merged.trade_date) >= pd.to_datetime(merged.sellable_from_trade_date)).all()
            and (pd.to_datetime(merged.trade_date) > pd.to_datetime(merged.entry_trade_date)).all()
        )
    else:
        t1_ok = True
    check("t1_guard_for_sell_orders", bool(t1_ok), f"sell_orders={len(sell_orders_df)}", checks)

    recompute_ok = True
    for _, row in closed.iterrows():
        entry = float(row.entry_price)
        exit_price = float(row.exit_price)
        position = float(row.position_pct)
        lot_ret = exit_price / entry - 1
        contrib = lot_ret * position
        recompute_ok = recompute_ok and abs(lot_ret - float(row.lot_return_pct)) < 1e-6
        recompute_ok = recompute_ok and abs(contrib - float(row.account_return_contribution_pct)) < 1e-6
    check("lot_return_recompute", bool(recompute_ok), f"closed_lots={len(closed)}", checks)

    case_ok = True
    for _, row in case_summary.iterrows():
        group = lots[lots.case_id == row.case_id]
        contrib = pd.to_numeric(group.account_return_contribution_pct, errors="coerce").fillna(0).sum()
        case_ok = case_ok and abs(contrib - float(row.account_return_closed_lots)) < 1e-6
        case_ok = case_ok and str(row.strict_baseline_return_ready_flag) in ["0", "False", "false"]
    check("case_summary_recompute", bool(case_ok), f"cases={len(case_summary)}", checks)

    account_ok = True
    account_direction_ok = True
    account_balance_ok = True
    final_open_ok = True
    sell_open_position_ok = True
    for case_id, group in account.groupby("case_id", sort=False):
        # Preserve ledger row order. Same-minute exits can contain multiple symbols;
        # re-sorting by symbol breaks the serial account state already recorded.
        sorted_group = group.copy()
        prev_cash = 1.0
        prev_open = 0.0
        for _, event in sorted_group.iterrows():
            cash = float(event.cash_pct_after_event)
            open_pos = float(event.open_position_pct_after_event)
            nav = float(event.account_nav_after_event)
            realized_delta = float(event.realized_return_delta)
            account_balance_ok = account_balance_ok and abs(nav - (cash + open_pos)) < 1e-6
            if event.action == "BUY":
                account_direction_ok = account_direction_ok and cash < prev_cash and open_pos > prev_open and abs(realized_delta) < 1e-9
            elif event.action == "SELL":
                account_direction_ok = account_direction_ok and cash > prev_cash and open_pos < prev_open
                sell_open_position_ok = sell_open_position_ok and open_pos >= -1e-9
            prev_cash = cash
            prev_open = open_pos
        last = sorted_group.iloc[-1]
        expected_nav = 1.0 + pd.to_numeric(
            lots[lots.case_id == case_id].account_return_contribution_pct,
            errors="coerce",
        ).fillna(0).sum()
        account_ok = account_ok and abs(float(last.account_nav_after_event) - expected_nav) < 1e-6
        expected_open = pd.to_numeric(
            lots[(lots.case_id == case_id) & (lots.lot_status != "CLOSED_BY_AI_SELL")].position_pct,
            errors="coerce",
        ).fillna(0).sum()
        final_open_ok = final_open_ok and abs(float(last.open_position_pct_after_event) - expected_open) < 1e-6
    check("daily_account_ledger_recompute", bool(account_ok), f"event_rows={len(account)}", checks)
    check("account_cash_position_direction", bool(account_direction_ok), "per-case BUY cash down/open up; SELL cash up/open down", checks)
    check("account_nav_equals_cash_plus_open_position", bool(account_balance_ok), "nav equals cash plus open position after each event", checks)
    check("account_final_open_position_matches_unclosed_lots", bool(final_open_ok), "final open position equals unresolved lot position sum per case", checks)
    check("sell_reduces_open_position_without_negative_open", bool(sell_open_position_ok), "SELL events reduce open position and never leave negative open position", checks)

    lookahead_ok = True
    for df in [order_ledger, decisions, sell_decisions]:
        if "lookahead_violation_flag" in df.columns:
            lookahead_ok = lookahead_ok and as_bool_false(df.lookahead_violation_flag)
    check("lookahead_flags_zero", bool(lookahead_ok), "order/decision/sell_decision flags checked", checks)

    chart_rows = []
    image_ok = True
    for _, row in image_manifest.iterrows():
        path = ROOT / str(row.path)
        exists = path.exists()
        actual_hash = sha256_file(path) if exists else ""
        hash_ok = exists and actual_hash == str(row.sha256)
        image_ok = image_ok and hash_ok
        chart_rows.append(
            {
                "case_id": row.case_id,
                "symbol": row.symbol,
                "chart_role": row.chart_role,
                "path": row.path,
                "exists": str(exists),
                "sha256_match": str(hash_ok),
                "decision_time": row.decision_time,
            }
        )
    chart_audit = pd.DataFrame(chart_rows)
    chart_audit.to_csv(ROOT / "chart_evidence_audit.csv", index=False, encoding="utf-8-sig")
    role_counts = image_manifest.chart_role.value_counts().to_dict()
    check("image_manifest_hash_match", bool(image_ok), f"images={len(image_manifest)}, roles={role_counts}", checks)
    open_selected_count = int((selected.market_gate_status == "MKT_GATE_OPEN_PREV_DAY_UP_3000").sum())
    expected_counts = {
        "candidate_daily_100d_decision_view": len(selected),
        "entry_1m_morning_review_view": open_selected_count,
        "entry_1m_late_review_view": open_selected_count,
        "entry_1m_buy_decision_view": buy_orders,
        "exit_daily_signal_review_view": len(lots),
        "exit_1m_sell_decision_view": sell_orders,
    }
    check(
        "image_manifest_expected_role_counts",
        all(int(role_counts.get(role, 0)) == expected for role, expected in expected_counts.items())
        and len(image_manifest) == sum(expected_counts.values()),
        f"actual={role_counts}; expected={expected_counts}",
        checks,
    )

    link_rows = []
    links_ok = True
    board_paths = [ROOT / "case_image_board.md", *sorted((ROOT / "cases").glob("*/case_image_board.md"))]
    for board in board_paths:
        if not board.exists():
            links_ok = False
            link_rows.append({"board": board.relative_to(ROOT).as_posix(), "target": "", "exists": "False"})
            continue
        for target in local_markdown_links(board):
            exists = target.exists()
            links_ok = links_ok and exists
            rel_target = target.relative_to(ROOT).as_posix() if target.exists() and ROOT in target.parents else str(target)
            link_rows.append({"board": board.relative_to(ROOT).as_posix(), "target": rel_target, "exists": str(exists)})
    pd.DataFrame(link_rows).to_csv(ROOT / "link_evidence_audit.csv", index=False, encoding="utf-8-sig")
    check("case_image_board_links_reachable", bool(links_ok), f"boards={len(board_paths)}, links={len(link_rows)}", checks)

    board_status_ok, board_status_detail = board_status_matches_summary(case_index, lots)
    check("BOARD_STATUS_MATCHES_SUMMARY", bool(board_status_ok), board_status_detail, checks)

    summary_files = [ROOT / "expand_run_config.json", ROOT / "expand_run_config.md", ROOT / "run_config.json", ROOT / "run_config.md"]
    check(
        "expand_run_config_files_present",
        all(p.exists() for p in summary_files),
        ",".join(p.name for p in summary_files),
        checks,
    )

    check(
        "return_stat_not_ready_boundary",
        (case_summary.strict_baseline_return_ready_flag.astype(str).isin(["0", "False", "false"])).all(),
        "RETURN_STAT_READY is intentionally false for this expansion execution package",
        checks,
    )

    checks_df = pd.DataFrame(checks)
    checks_df.to_csv(ROOT / "self_check_items.csv", index=False, encoding="utf-8-sig")
    fail_count = int((checks_df.status == "FAIL").sum())
    summary = {
        "schema_version": "1.0",
        "run_id": RUN_ID,
        "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "stage": "EXPAND_30_EXECUTION_SELF_CHECK_DONE",
        "overall_status": "PASS_FOR_EXPAND_30_EXECUTION_REVIEW_READY" if fail_count == 0 else "FAIL",
        "fail_count": fail_count,
        "check_count": int(len(checks_df)),
        "case_count": int(len(case_index)),
        "selected_candidate_rows": int(len(selected)),
        "order_counts": {k: int(v) for k, v in action_counts.items()},
        "lot_status_counts": {k: int(v) for k, v in lots.lot_status.value_counts().to_dict().items()},
        "image_role_counts": {k: int(v) for k, v in role_counts.items()},
        "closed_lot_account_return_sum": float(pd.to_numeric(lots.account_return_contribution_pct, errors="coerce").fillna(0).sum()),
        "strict_baseline_return_ready_flag": False,
        "boundary": (
            "Machine self-check passed only for the current 30-case controlled expansion execution package. "
            "Execution audit is still required and this is not a full 2023-2026 baseline conclusion."
        ),
        "artifacts": {
            "self_check_items.csv": {"size": (ROOT / "self_check_items.csv").stat().st_size, "sha256": sha256_file(ROOT / "self_check_items.csv")},
            "chart_evidence_audit.csv": {"size": (ROOT / "chart_evidence_audit.csv").stat().st_size, "sha256": sha256_file(ROOT / "chart_evidence_audit.csv")},
            "link_evidence_audit.csv": {"size": (ROOT / "link_evidence_audit.csv").stat().st_size, "sha256": sha256_file(ROOT / "link_evidence_audit.csv")},
        },
    }
    (ROOT / "self_check.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    (ROOT / "self_check.md").write_text(
        "\n".join(
            [
                "# self_check",
                "",
                f"run_id：`{RUN_ID}`",
                f"状态：`{summary['overall_status']}`",
                f"检查项：{len(checks_df)}，失败：{fail_count}",
                "",
                "关键读数：",
                f"- 案例日：{summary['case_count']}",
                f"- 选中候选：{summary['selected_candidate_rows']}",
                f"- BUY：{buy_orders}",
                f"- SELL：{sell_orders}",
                f"- 已闭合 lot：{len(closed)}",
                f"- 未闭合 / 边界 lot：{len(unresolved)}",
                f"- 图片：{len(image_manifest)}",
                f"- 闭合 lot 账户贡献合计：{summary['closed_lot_account_return_sum']:.4%}",
                "",
                "边界：当前只通过扩样执行包机器自检；执行审核未通过前不得引用为完整 baseline 收益、成功率、胜率、回撤或策略有效性结论。",
                "",
            ]
        ),
        encoding="utf-8",
    )


if __name__ == "__main__":
    main()