#!/usr/bin/env python3
"""Replay the frozen daily ensemble through downloaded 2000-2025 minute bars.

The daily alpha members are frozen from the prior calibration/validation study.
This script does not re-select them with final-holdout data. It uses minute bars
for two narrower purposes:

1. replace the idealized daily-open fill with time/volume participation policies;
2. test a lagged broad-index risk overlay selected before the final holdout.

No broker is connected and no order is submitted.
"""

from __future__ import annotations

import argparse
import collections
import csv
import datetime as dt
import io
import json
import math
import re
import statistics
import sys
import zipfile
from array import array
from dataclasses import dataclass
from pathlib import Path


PORTAL_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PORTAL_ROOT / ".python-deps"))
sys.path.insert(0, str(Path(__file__).resolve().parent))

from local_archive_streaming_optimizer import (  # noqa: E402
    Config,
    Simulation,
    SymbolState,
    annualized_volatility,
    date_text,
    is_a_share,
    limit_fraction,
    metric,
    safe_float,
    score_metric,
    tar_parquet_rows,
)


QUANT_ROOT = Path("/Users/zhuboyu/Desktop/龙门计划/投资/Quant")
DEFAULT_STOCK_MINUTE = QUANT_ROOT / "1分钟_沪深_按年汇总"
DEFAULT_BEIJING_MINUTE = QUANT_ROOT / "1分钟_京市_按年汇总"
DEFAULT_INDEX_MINUTE = QUANT_ROOT / "1分钟_指数_按年汇总"
DEFAULT_TUSHARE = QUANT_ROOT / "tushare_15000_history_by_api_packages_20260627"
DEFAULT_FROZEN_RESULT = PORTAL_ROOT / "research/results/local-archive-streaming-optimizer-v1.json"


RANGES = {
    "calibration": {"start": "2000-01-01", "end": "2013-12-31"},
    "validation": {"start": "2014-01-01", "end": "2021-12-31"},
    "finalHoldout": {"start": "2022-01-01", "end": "2025-12-31"},
    "fullResearch": {"start": "2000-01-01", "end": "2025-12-31"},
}


@dataclass(frozen=True)
class ExecutionPolicy:
    identifier: str
    start_hhmm: int
    window_minutes: int
    participation: float
    eligible: bool = True


POLICIES = (
    ExecutionPolicy("THEORETICAL_DAILY_OPEN", 930, 1, 1.0, False),
    ExecutionPolicy("OPENING_BAR_POV_2PCT_5M", 930, 5, 0.02),
    ExecutionPolicy("LATENCY_1M_POV_2PCT_5M", 931, 5, 0.02),
    ExecutionPolicy("LATENCY_1M_POV_1PCT_15M", 931, 15, 0.01),
    ExecutionPolicy("LATENCY_5M_POV_1PCT_15M", 935, 15, 0.01),
)


def percentile(values, q):
    if not values:
        return None
    ordered = sorted(values)
    position = (len(ordered) - 1) * q
    lower = int(math.floor(position))
    upper = int(math.ceil(position))
    if lower == upper:
        return ordered[lower]
    return ordered[lower] * (upper - position) + ordered[upper] * (position - lower)


def minute_name(code: str, year: int) -> str:
    symbol, exchange = code.split(".")
    prefix = "bj" if exchange == "BJ" else "sh" if exchange == "SH" else "sz"
    return f"{prefix}{symbol}_{year}.txt"


def selected_rankings(states: dict[str, SymbolState], configs: list[Config]):
    lookbacks = sorted({config.lookback for config in configs})
    breadth = {lookback: [0, 0] for lookback in lookbacks}
    items_by_lookback = {lookback: [] for lookback in lookbacks}
    for code, state in states.items():
        if state.days < 120 or len(state.closes) <= 60:
            continue
        current = state.closes[-1]
        long_past = state.closes[0] if len(state.closes) < 201 else state.closes[-201]
        if min(current, long_past) <= 0:
            continue
        average_amount = statistics.mean(state.amounts) if state.amounts else 0
        if average_amount < 50_000:
            continue
        vol = annualized_volatility(state.returns)
        if not 0.02 <= vol <= 1.20:
            continue
        if state.total_market_value and state.total_market_value < 200_000:
            continue
        if state.turnover_rate > 30:
            continue
        long_momentum = current / long_past - 1
        for lookback in lookbacks:
            if len(state.closes) <= lookback:
                continue
            past = state.closes[-lookback - 1]
            if past <= 0:
                continue
            momentum = current / past - 1
            breadth[lookback][1] += 1
            breadth[lookback][0] += long_momentum > 0
            if long_momentum <= 0 or momentum <= -0.12:
                continue
            value = -math.log(max(0.2, min(20, state.pb))) if state.pb > 0 else 0.0
            earnings_yield = 1 / state.pe_ttm if 0 < state.pe_ttm < 200 else 0.0
            dividend = max(0.0, min(15.0, state.dividend_yield)) / 100
            items_by_lookback[lookback].append({
                "code": code,
                "vol": vol,
                "long": long_momentum,
                "value": value + 3 * earnings_yield,
                "dividend": dividend,
                "lowvol": -vol,
            })

    output = {}
    for lookback, items in items_by_lookback.items():
        def zscores(name):
            values = [item[name] for item in items]
            average = statistics.mean(values) if values else 0
            deviation = statistics.pstdev(values) if len(values) > 1 else 1
            return [(value - average) / max(1e-9, deviation) for value in values]

        factor_z = {name: zscores(name) for name in ("long", "value", "dividend", "lowvol")}
        ranked = []
        for index, item in enumerate(items):
            score = (
                0.35 * factor_z["value"][index]
                + 0.20 * factor_z["dividend"][index]
                + 0.30 * factor_z["lowvol"][index]
                + 0.15 * factor_z["long"][index]
            ) / max(0.08, item["vol"])
            ranked.append((score, item["code"], item["vol"]))
        output[lookback] = (
            breadth[lookback][0] / max(1, breadth[lookback][1]),
            sorted(ranked, reverse=True)[:200],
        )
    return output


def update_and_record(
    sim: Simulation,
    rows: dict[str, dict],
    date: str,
    date_index: int,
    cost_bps: float,
    member_index: int,
    events: list[dict],
):
    gross_return = 0.0
    factors = {}
    for code, weight in sim.active.items():
        row = rows.get(code)
        factor = row["openFactor"] if row else 1.0
        factors[code] = factor
        gross_return += weight * (factor - 1)
    if sim.active and 1 + gross_return > 0:
        sim.active = {code: weight * factors.get(code, 1.0) / (1 + gross_return) for code, weight in sim.active.items()}

    turnover = 0.0
    if sim.pending is not None:
        executed = dict(sim.active)
        for code in set(sim.active) | set(sim.pending):
            old = sim.active.get(code, 0.0)
            target = sim.pending.get(code, 0.0)
            row = rows.get(code)
            blocked = row is None
            if row is not None:
                threshold = limit_fraction(code, date) * 0.985
                open_gap = row["rawOpen"] / row["preClose"] - 1 if row["preClose"] > 0 else 0
                blocked = (target > old and open_gap >= threshold) or (target < old and open_gap <= -threshold)
            if blocked:
                sim.blocked_orders += 1
                continue
            delta = target - old
            if abs(delta) > 1e-9:
                events.append({
                    "member": member_index,
                    "date": date,
                    "code": code,
                    "deltaWeight": delta,
                    "rawOpen": row["rawOpen"],
                })
            if target > 1e-9:
                executed[code] = target
            else:
                executed.pop(code, None)
        for code in set(sim.active) | set(executed):
            turnover += abs(sim.active.get(code, 0.0) - executed.get(code, 0.0))
        sim.active = executed
        sim.pending = None

    net_return = max(-0.25, gross_return - turnover * cost_bps / 10_000)
    sim.turnover += turnover
    sim.returns.append(net_return)
    sim.daily_turnover.append(turnover)
    sim.equity *= 1 + net_return
    sim.peak = max(sim.peak, sim.equity)
    drawdown = sim.equity / sim.peak - 1
    if not sim.cooldown_armed and date_index > sim.cooldown_until:
        sim.peak = sim.equity
        sim.cooldown_armed = True
        drawdown = 0.0
    if sim.cooldown_armed and drawdown <= -0.08:
        sim.cooldown_until = date_index + 20
        sim.cooldown_armed = False


def replay_frozen_ensemble(tushare_root: Path, frozen_result: Path, cost_bps: float):
    frozen = json.loads(frozen_result.read_text(encoding="utf-8"))
    ids = frozen["validationSelectedEnsemble"]["memberIds"]
    config_lookup = {item["id"]: Config(**item["config"]) for item in frozen["topValidationCandidates"]}
    configs = [config_lookup[identifier] for identifier in ids]
    simulations = [Simulation(config) for config in configs]
    states: dict[str, SymbolState] = {}
    dates = []
    events = []
    total_rows = 0
    daily_tar = tushare_root / "daily" / "daily.tar.gz"
    daily_basic_tar = tushare_root / "daily_basic" / "daily_basic.tar.gz"

    for date_index, (raw_date, columns, basic_columns) in enumerate(tar_parquet_rows(daily_tar, daily_basic_tar)):
        date = date_text(raw_date)
        if date > "2025-12-31":
            break
        dates.append(date)
        rows = {}
        basic_by_code = {}
        if basic_columns:
            for basic_index, basic_code in enumerate(basic_columns["ts_code"]):
                basic_by_code[basic_code] = {
                    "pe_ttm": safe_float(basic_columns["pe_ttm"][basic_index]),
                    "pb": safe_float(basic_columns["pb"][basic_index]),
                    "dv_ttm": safe_float(basic_columns["dv_ttm"][basic_index]),
                    "total_mv": safe_float(basic_columns["total_mv"][basic_index]),
                    "turnover_rate": safe_float(basic_columns["turnover_rate"][basic_index]),
                }
        for index, code in enumerate(columns["ts_code"]):
            if not is_a_share(code):
                continue
            raw_open = safe_float(columns["open"][index])
            raw_close = safe_float(columns["close"][index])
            pre_close = safe_float(columns["pre_close"][index])
            pct_change = safe_float(columns["pct_chg"][index]) / 100
            amount = safe_float(columns["amount"][index])
            state = states.setdefault(code, SymbolState())
            previous_normalized = state.normalized_close
            normalized_open = previous_normalized * raw_open / pre_close if previous_normalized > 0 and pre_close > 0 else raw_open
            normalized_close = previous_normalized * (1 + pct_change) if previous_normalized > 0 else raw_close
            prior_open = getattr(state, "last_open", 0.0)
            rows[code] = {
                "rawOpen": raw_open,
                "preClose": pre_close,
                "openFactor": normalized_open / prior_open if prior_open > 0 and normalized_open > 0 else 1.0,
            }
            state.last_open = normalized_open
            state.normalized_close = normalized_close
            state.raw_close = raw_close
            state.raw_date = date
            state.days += 1
            state.closes.append(normalized_close)
            state.returns.append(pct_change)
            state.amounts.append(amount)
            basic = basic_by_code.get(code)
            if basic:
                state.pe_ttm = basic["pe_ttm"]
                state.pb = basic["pb"]
                state.dividend_yield = basic["dv_ttm"]
                state.total_market_value = basic["total_mv"]
                state.turnover_rate = basic["turnover_rate"]
            total_rows += 1

        for member_index, sim in enumerate(simulations):
            update_and_record(sim, rows, date, date_index, cost_bps, member_index, events)

        if date_index >= 201 and date_index % 5 == 0:
            rankings = selected_rankings(states, configs)
            for config, sim in zip(configs, simulations):
                if date_index % config.rebalance_days:
                    continue
                breadth, ranking = rankings[config.lookback]
                if date_index <= sim.cooldown_until or breadth < config.minimum_breadth:
                    sim.pending = {}
                    continue
                selected = ranking[: config.top_k]
                exposure = config.exposure * (0.5 if sim.equity / sim.peak - 1 <= -0.04 else 1.0)
                if not selected:
                    sim.pending = {}
                    continue
                inverse_vol = [1 / max(0.08, item[2]) for item in selected]
                divisor = sum(inverse_vol)
                sim.pending = {item[1]: min(0.05, exposure * inverse_vol[index] / divisor) for index, item in enumerate(selected)}
        if (date_index + 1) % 750 == 0:
            print(f"daily replay: {date_index + 1:,} sessions, {total_rows:,} rows, {len(events):,} orders", flush=True)

    base_returns = array("d", (
        statistics.mean(float(sim.returns[index]) for sim in simulations)
        for index in range(len(dates))
    ))
    base_turnover = array("d", (
        statistics.mean(float(sim.daily_turnover[index]) for sim in simulations)
        for index in range(len(dates))
    ))
    return {
        "dates": dates,
        "configs": configs,
        "simulations": simulations,
        "events": events,
        "baseReturns": base_returns,
        "baseTurnover": base_turnover,
        "rows": total_rows,
    }


def bars_for_dates(zf: zipfile.ZipFile, member: str, target_dates: set[str]):
    output = collections.defaultdict(list)
    with zf.open(member) as raw, io.TextIOWrapper(raw, encoding="utf-8-sig", errors="replace", newline="") as text:
        reader = csv.reader(text)
        next(reader, None)
        for row in reader:
            if len(row) < 8:
                continue
            raw_date = row[1]
            if raw_date not in target_dates:
                continue
            hhmm = int(row[2]) if row[2].isdigit() else 0
            if not 930 <= hhmm <= 1005:
                continue
            values = [safe_float(value) for value in row[3:8]]
            output[raw_date].append((hhmm, *values))
    return output


def policy_cost(event, bars, policy: ExecutionPolicy, capital: float):
    weight = abs(event["deltaWeight"])
    side = 1 if event["deltaWeight"] > 0 else -1
    if policy.identifier == "THEORETICAL_DAILY_OPEN":
        return weight * 0.0015, 1.0, False
    if not bars:
        return weight * 0.0040, 0.0, True
    daily_open = event["rawOpen"]
    minute_open = bars[0][1]
    if min(daily_open, minute_open) <= 0 or abs(minute_open / daily_open - 1) > 0.05:
        return weight * 0.0040, 0.0, True
    eligible = [bar for bar in bars if bar[0] >= policy.start_hhmm][: policy.window_minutes]
    if not eligible:
        return weight * 0.0040, 0.0, True
    requested = max(1.0, weight * capital)
    remaining = requested
    filled_value = 0.0
    filled_notional = 0.0
    total_market_notional = 0.0
    for _, open_, high, low, close, volume in eligible:
        typical = max(0.0001, (open_ + high + low + close) / 4)
        market_notional = typical * max(0.0, volume) * 100
        total_market_notional += market_notional
        capacity = market_notional * policy.participation
        fill = min(remaining, capacity)
        filled_value += fill * typical
        filled_notional += fill
        remaining -= fill
        if remaining <= 1e-6:
            break
    last_price = max(0.0001, eligible[-1][4])
    if remaining > 0:
        forced_price = last_price * (1 + side * 0.0020)
        filled_value += remaining * forced_price
        filled_notional += remaining
    execution_price = filled_value / max(1e-9, filled_notional)
    realized_participation = requested / max(requested, total_market_notional)
    impact_bps = min(20.0, 2.0 + 25.0 * math.sqrt(max(0.0, realized_participation)))
    adverse_move = side * (execution_price / daily_open - 1)
    all_in_adverse = adverse_move + (3.0 + impact_bps) / 10_000
    fill_ratio = min(1.0, (requested - max(0.0, remaining)) / requested)
    return weight * all_in_adverse, fill_ratio, False


def apply_minute_execution(replay, stock_root: Path, beijing_root: Path, cost_bps: float, capital: float):
    dates = replay["dates"]
    date_index = {date: index for index, date in enumerate(dates)}
    member_count = len(replay["simulations"])
    policy_costs = {
        policy.identifier: [array("d", [0.0] * len(dates)) for _ in range(member_count)]
        for policy in POLICIES
    }
    policy_fill = {policy.identifier: [] for policy in POLICIES}
    policy_fallbacks = collections.Counter()
    events_by_year_code = collections.defaultdict(lambda: collections.defaultdict(list))
    for event in replay["events"]:
        if event["date"] > "2025-12-31":
            continue
        events_by_year_code[int(event["date"][:4])][event["code"]].append(event)

    minute_open_differences = []
    matched_events = 0
    missing_members = 0
    archives_read = 0
    for year, by_code in sorted(events_by_year_code.items()):
        roots = {"BJ": beijing_root, "SH": stock_root, "SZ": stock_root}
        archives = {}
        for exchange, root in roots.items():
            path = root / f"{year}_1min.zip"
            if path.exists() and path not in archives:
                archives[path] = zipfile.ZipFile(path)
        name_sets = {path: set(zf.namelist()) for path, zf in archives.items()}
        archives_read += len(archives)
        for code, events in by_code.items():
            exchange = code.split(".")[1]
            path = roots[exchange] / f"{year}_1min.zip"
            zf = archives.get(path)
            member = minute_name(code, year)
            if zf is None or member not in name_sets[path]:
                missing_members += 1
                by_date = {}
            else:
                by_date = bars_for_dates(zf, member, {event["date"].replace("-", "") for event in events})
            for event in events:
                raw_date = event["date"].replace("-", "")
                bars = by_date.get(raw_date, [])
                if bars and event["rawOpen"] > 0:
                    minute_open_differences.append(abs(bars[0][1] / event["rawOpen"] - 1))
                    matched_events += 1
                index = date_index[event["date"]]
                for policy in POLICIES:
                    cost, fill, fallback = policy_cost(event, bars, policy, capital)
                    policy_costs[policy.identifier][event["member"]][index] += cost
                    policy_fill[policy.identifier].append(fill)
                    if fallback:
                        policy_fallbacks[policy.identifier] += 1
        for zf in archives.values():
            zf.close()
        print(f"minute execution: {year}, {len(by_code):,} symbols, {sum(len(v) for v in by_code.values()):,} orders", flush=True)

    results = []
    return_series = {}
    for policy in POLICIES:
        member_returns = []
        for member_index, sim in enumerate(replay["simulations"]):
            values = array("d", (
                float(base) + float(turnover) * cost_bps / 10_000 - policy_costs[policy.identifier][member_index][index]
                for index, (base, turnover) in enumerate(zip(sim.returns, sim.daily_turnover))
            ))
            member_returns.append(values)
        ensemble = array("d", (
            statistics.mean(float(values[index]) for values in member_returns)
            for index in range(len(dates))
        ))
        return_series[policy.identifier] = ensemble
        item = {
            "id": policy.identifier,
            "startHHMM": policy.start_hhmm,
            "windowMinutes": policy.window_minutes,
            "participation": policy.participation,
            "eligible": policy.eligible,
            "meanImmediateFillRatio": round(statistics.mean(policy_fill[policy.identifier]), 6) if policy_fill[policy.identifier] else 0,
            "fallbackEvents": policy_fallbacks[policy.identifier],
        }
        for name, span in RANGES.items():
            item[name] = metric(ensemble, dates, span["start"], span["end"])
        results.append(item)

    eligible = [item for item in results if item["eligible"]]
    calibration_top = sorted(eligible, key=lambda item: score_metric(item["calibration"]), reverse=True)[:3]
    selected = max(calibration_top, key=lambda item: score_metric(item["validation"]))
    audit = {
        "archivesRead": archives_read,
        "ordersRequested": sum(len(value) for by_code in events_by_year_code.values() for value in by_code.values()),
        "ordersWithMinuteBars": matched_events,
        "missingSymbolYearMembers": missing_members,
        "minuteVsDailyOpenMedianAbsoluteDifference": round(percentile(minute_open_differences, 0.50) or 0, 8),
        "minuteVsDailyOpenP95AbsoluteDifference": round(percentile(minute_open_differences, 0.95) or 0, 8),
        "volumeUnitAssumption": "VOL_IS_100_SHARE_LOTS_FOR_STOCK_CAPACITY_PROXY",
        "level2Available": False,
    }
    return results, selected, return_series, audit


def load_index_features(index_root: Path):
    features = {}
    coverage = collections.Counter()
    archive_audit = []
    for path in sorted(index_root.glob("*_1min.zip")):
        match = re.match(r"(\d{4})_1min\.zip", path.name)
        if not match:
            continue
        year = int(match.group(1))
        with zipfile.ZipFile(path) as zf:
            names = set(zf.namelist())
            candidates = [f"000300_{year}.txt", f"000001_{year}.txt", f"399001_{year}.txt"]
            member = next((name for name in candidates if name in names), None)
            archive_audit.append({"year": year, "members": len(names), "broadIndexMember": member})
            if member is None:
                continue
            by_date = collections.defaultdict(list)
            with zf.open(member) as raw, io.TextIOWrapper(raw, encoding="utf-8-sig", errors="replace", newline="") as text:
                reader = csv.reader(text)
                next(reader, None)
                for row in reader:
                    if len(row) < 8:
                        continue
                    by_date[row[1]].append((int(row[2]), *(safe_float(value) for value in row[3:8])))
            for raw_date, bars in by_date.items():
                bars.sort()
                closes = [bar[4] for bar in bars if bar[4] > 0]
                if len(closes) < 100:
                    continue
                realized = math.sqrt(sum(math.log(closes[index] / closes[index - 1]) ** 2 for index in range(1, len(closes)) if closes[index - 1] > 0))
                volume_total = sum(max(0.0, bar[5]) for bar in bars)
                vwap = sum(((bar[1] + bar[2] + bar[3] + bar[4]) / 4) * max(0.0, bar[5]) for bar in bars) / max(1e-9, volume_total)
                date = date_text(raw_date)
                features[date] = {
                    "date": date,
                    "source": member.split("_")[0],
                    "open": bars[0][1],
                    "close": bars[-1][4],
                    "vwap": vwap,
                    "intradayReturn": bars[-1][4] / bars[0][1] - 1 if bars[0][1] > 0 else 0,
                    "realizedVol": realized,
                }
                coverage[member.split("_")[0]] += 1
        print(f"index features: {year}, {sum(1 for date in features if date.startswith(str(year))):,} sessions", flush=True)
    return features, {"sessions": len(features), "bySource": dict(coverage), "archives": archive_audit}


def overlay_series(base_returns, dates, features, config):
    ordered_features = [features[date] for date in sorted(features)]
    signals = {}
    closes = collections.deque(maxlen=config["trendLookback"])
    vols = collections.deque(maxlen=config["volWindow"])
    for item in ordered_features:
        closes.append(item["close"])
        vols.append(item["realizedVol"])
        trend_weak = len(closes) == closes.maxlen and item["close"] < statistics.mean(closes)
        vol_median = statistics.median(vols) if len(vols) >= max(10, vols.maxlen // 2) else item["realizedVol"]
        vol_high = item["realizedVol"] > vol_median * config["volRatio"]
        multiplier = 1.0
        if trend_weak:
            multiplier = min(multiplier, config["weakMultiplier"])
        if vol_high:
            multiplier = min(multiplier, config["highVolMultiplier"])
        signals[item["date"]] = multiplier
    values = array("d")
    previous_multiplier = 1.0
    for index, value in enumerate(base_returns):
        # Two-session lag: a close-known signal cannot alter an already-running
        # open-to-open return. This is deliberately slower than a live close order.
        source_date = dates[index - 2] if index >= 2 else None
        multiplier = signals.get(source_date, 1.0) if source_date else 1.0
        change_cost = abs(multiplier - previous_multiplier) * 0.0002
        values.append(float(value) * multiplier - change_cost)
        previous_multiplier = multiplier
    return values


def optimize_overlay(base_returns, dates, features):
    configs = []
    for trend in (20, 60, 120):
        for weak in (0.25, 0.50, 0.75):
            for vol_window in (20, 60):
                for vol_ratio in (1.0, 1.25, 1.50):
                    for high_vol in (0.50, 0.75, 1.0):
                        configs.append({
                            "trendLookback": trend,
                            "weakMultiplier": weak,
                            "volWindow": vol_window,
                            "volRatio": vol_ratio,
                            "highVolMultiplier": high_vol,
                        })
    evaluated = []
    series_by_id = {}
    for config in configs:
        identifier = f"IDX-L{config['trendLookback']}-W{config['weakMultiplier']}-V{config['volWindow']}x{config['volRatio']}-H{config['highVolMultiplier']}"
        series = overlay_series(base_returns, dates, features, config)
        series_by_id[identifier] = series
        item = {"id": identifier, "config": config}
        for name, span in RANGES.items():
            item[name] = metric(series, dates, span["start"], span["end"])
        evaluated.append(item)
    calibration_top = sorted(evaluated, key=lambda item: score_metric(item["calibration"]), reverse=True)[:30]
    selected = max(calibration_top, key=lambda item: score_metric(item["validation"]))
    return evaluated, selected, series_by_id[selected["id"]]


def inventory(root: Path, expected_years: set[int]):
    files = sorted(root.glob("*_1min.zip"))
    years = {int(path.name[:4]) for path in files if re.match(r"\d{4}_1min\.zip", path.name)}
    members = 0
    uncompressed = 0
    bad_archives = []
    for path in files:
        try:
            with zipfile.ZipFile(path) as zf:
                members += len(zf.infolist())
                uncompressed += sum(info.file_size for info in zf.infolist())
        except zipfile.BadZipFile:
            bad_archives.append(str(path))
    return {
        "path": str(root),
        "archives": len(files),
        "archiveBytes": sum(path.stat().st_size for path in files),
        "uncompressedBytes": uncompressed,
        "members": members,
        "years": sorted(years),
        "missingYears": sorted(expected_years - years),
        "unexpectedYears": sorted(years - expected_years),
        "badCentralDirectories": bad_archives,
    }


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--stock-minute", type=Path, default=DEFAULT_STOCK_MINUTE)
    parser.add_argument("--beijing-minute", type=Path, default=DEFAULT_BEIJING_MINUTE)
    parser.add_argument("--index-minute", type=Path, default=DEFAULT_INDEX_MINUTE)
    parser.add_argument("--tushare-root", type=Path, default=DEFAULT_TUSHARE)
    parser.add_argument("--frozen-result", type=Path, default=DEFAULT_FROZEN_RESULT)
    parser.add_argument("--result", type=Path, required=True)
    parser.add_argument("--report", type=Path, required=True)
    parser.add_argument("--cost-bps", type=float, default=15.0)
    parser.add_argument("--capital", type=float, default=1_000_000.0)
    args = parser.parse_args()

    inventories = {
        "stock": inventory(args.stock_minute, set(range(2000, 2026))),
        "beijing": inventory(args.beijing_minute, set(range(2020, 2026))),
        "index": inventory(args.index_minute, set(range(2000, 2026))),
    }
    if any(item["missingYears"] or item["badCentralDirectories"] for item in inventories.values()):
        raise RuntimeError(f"minute archive inventory failed: {inventories}")

    replay = replay_frozen_ensemble(args.tushare_root, args.frozen_result, args.cost_bps)
    execution_results, execution_selected, execution_series, minute_audit = apply_minute_execution(
        replay, args.stock_minute, args.beijing_minute, args.cost_bps, args.capital
    )
    index_features, index_audit = load_index_features(args.index_minute)
    overlay_results, overlay_selected, combined_series = optimize_overlay(
        execution_series[execution_selected["id"]], replay["dates"], index_features
    )
    base_metrics = {name: metric(replay["baseReturns"], replay["dates"], span["start"], span["end"]) for name, span in RANGES.items()}
    combined_metrics = {name: metric(combined_series, replay["dates"], span["start"], span["end"]) for name, span in RANGES.items()}
    baseline_holdout = base_metrics["finalHoldout"]
    combined_holdout = combined_metrics["finalHoldout"]
    holdout_gate = (
        combined_holdout["annualizedReturn"] >= baseline_holdout["annualizedReturn"]
        and combined_holdout["maxDrawdown"] >= baseline_holdout["maxDrawdown"]
        and combined_holdout["sharpe"] >= baseline_holdout["sharpe"]
    )

    result = {
        "version": "historical-minute-execution-optimizer-v1",
        "generatedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
        "mode": "RESEARCH_ONLY_REAL_ORDERS_LOCKED",
        "inventories": inventories,
        "dailyReplay": {
            "sessions": len(replay["dates"]),
            "rows": replay["rows"],
            "start": replay["dates"][0],
            "end": replay["dates"][-1],
            "frozenMembers": [config.identifier for config in replay["configs"]],
            "orders": len(replay["events"]),
            "selectionRule": "daily alpha members frozen before this minute study",
            "baseline": base_metrics,
        },
        "minuteAudit": minute_audit,
        "executionPolicies": execution_results,
        "selectedExecutionPolicy": execution_selected,
        "indexAudit": index_audit,
        "overlayConfigsEvaluated": len(overlay_results),
        "selectedIndexOverlay": overlay_selected,
        "combined": combined_metrics,
        "gates": {
            "minuteDataAcceptedForResearch": minute_audit["ordersWithMinuteBars"] > 0 and minute_audit["minuteVsDailyOpenP95AbsoluteDifference"] < 0.01,
            "combinedImprovesAllHoldoutMetrics": holdout_gate,
            "allowedForRealCapital": False,
            "blockers": [
                "downloaded files do not contain ETF minute histories",
                "bar data cannot reproduce Level-2 queue priority or cancel/fill races",
                "no broker execution reports or IOPV feed",
                "requires forward shadow trading and explicit human approval",
            ],
        },
        "limitations": [
            "stock VOL is treated as 100-share lots for a conservative capacity proxy",
            "minute OHLCV has no bid/ask, order-book, auction queue, order IDs, or partial-fill reports",
            "final holdout is used only for go/no-go evaluation and never for parameter selection",
            "no random-walk price forecast is used",
        ],
    }
    args.result.parent.mkdir(parents=True, exist_ok=True)
    args.result.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    pct = lambda value: f"{value * 100:.2f}%"
    total_gib = sum(item["archiveBytes"] for item in inventories.values()) / 1073741824
    report = [
        "# A股2000–2025历史分钟执行优化与样本外审计",
        "",
        f"- 三目录：{sum(item['archives'] for item in inventories.values())}个年度ZIP，压缩体积{total_gib:.2f} GiB；中央目录均可读、年度无缺失。",
        f"- 冻结日线集成：{len(replay['configs'])}个成员、{len(replay['dates']):,}个交易日、{replay['rows']:,}条日线；分钟研究前已经冻结，不使用最终留出重新挑选。",
        f"- 分钟撮合请求：{minute_audit['ordersRequested']:,}笔；匹配分钟正文{minute_audit['ordersWithMinuteBars']:,}笔；分钟首价与日线开盘价绝对差中位数{pct(minute_audit['minuteVsDailyOpenMedianAbsoluteDifference'])}、P95 {pct(minute_audit['minuteVsDailyOpenP95AbsoluteDifference'])}。",
        "- 选参区间：2000–2013校准、2014–2021验证；2022–2025最终留出不参与选参。",
        "",
        "## 分钟执行策略",
        "",
        f"- 验证期选择：`{execution_selected['id']}`，从{execution_selected['startHHMM']}起、{execution_selected['windowMinutes']}分钟窗口、单分钟成交额参与率上限{pct(execution_selected['participation'])}。",
        f"- 即时容量完成率均值：{pct(execution_selected['meanImmediateFillRatio'])}；不足部分使用保守强制成交与额外冲击惩罚，不把未成交当成零成本。",
        f"- 验证期：年化{pct(execution_selected['validation']['annualizedReturn'])}，Sharpe {execution_selected['validation']['sharpe']:.2f}，最大回撤{pct(execution_selected['validation']['maxDrawdown'])}。",
        f"- 最终留出：年化{pct(execution_selected['finalHoldout']['annualizedReturn'])}，Sharpe {execution_selected['finalHoldout']['sharpe']:.2f}，最大回撤{pct(execution_selected['finalHoldout']['maxDrawdown'])}。",
        "",
        "## 指数分钟风险覆盖层",
        "",
        f"- 从指数分钟正文形成{index_audit['sessions']:,}个交易日的收盘趋势与实现波动率，只使用收盘已知信息，并额外滞后两个交易日映射到开盘到开盘收益。",
        f"- 共评估{len(overlay_results)}组覆盖层，验证期选择：`{overlay_selected['id']}`。",
        f"- 组合验证期：年化{pct(combined_metrics['validation']['annualizedReturn'])}，Sharpe {combined_metrics['validation']['sharpe']:.2f}，最大回撤{pct(combined_metrics['validation']['maxDrawdown'])}。",
        f"- 组合最终留出：年化{pct(combined_holdout['annualizedReturn'])}，Sharpe {combined_holdout['sharpe']:.2f}，最大回撤{pct(combined_holdout['maxDrawdown'])}，日胜率{pct(combined_holdout['dailyWinRate'])}。",
        "",
        "## 与原理想开盘基线比较",
        "",
        f"- 原基线最终留出：年化{pct(baseline_holdout['annualizedReturn'])}，Sharpe {baseline_holdout['sharpe']:.2f}，最大回撤{pct(baseline_holdout['maxDrawdown'])}。",
        f"- 三项留出指标是否同时改善：`{holdout_gate}`。该判断只决定接受/拒绝，不反向调参。",
        "",
        "## 边界",
        "",
        "这些分钟数据显著改善了历史成交时点与容量建模，但仍不是Level-2：没有买卖盘口、队列优先级、订单编号、撤单竞态和券商成交回报；目录也没有ETF分钟正文。因此真实资金继续LOCKED，分钟结果只能升级影子执行和研究证据，不能直接授权实盘。没有使用随机游走预测股价。",
        "",
    ]
    args.report.parent.mkdir(parents=True, exist_ok=True)
    args.report.write_text("\n".join(report), encoding="utf-8")
    print(json.dumps({
        "inventories": inventories,
        "dailyReplay": result["dailyReplay"],
        "minuteAudit": minute_audit,
        "selectedExecutionPolicy": execution_selected,
        "selectedIndexOverlay": overlay_selected,
        "combined": combined_metrics,
        "gates": result["gates"],
    }, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()