#!/usr/bin/env python3
"""Capital allocation flow implementation internals."""

from __future__ import annotations

from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
import json
import logging

from core.modules.strategy.engines.analyzer import Analyzer
from core.modules.strategy.engines.shared.data_classes.strategy_settings.dict_view_settings import (
    StrategySettingsView,
)
from core.modules.strategy.engines.simulator.capital_allocation.data_classes.settings import (
    StrategyCapitalSimulatorSettings,
)
from core.modules.strategy.engines.shared.data_classes.market_profile_context import (
    MarketProfileContext,
)
from core.modules.strategy.engines.shared.data_classes.strategy_settings.simulation_settings import (
    StrategySimulationSettings,
)
from core.modules.strategy.engines.shared.helpers.skip_investment_when import (
    ROW_SKIP_REASON_KEY,
    should_skip_investment,
)
from core.modules.strategy.engines.shared.helpers.participation import (
    apply_participation_to_shares,
    row_buy_bar_volume,
    row_sell_bar_volume,
)
from core.modules.strategy.engines.shared.helpers.tradability import (
    should_skip_buy,
    should_skip_sell,
)
from core.modules.strategy.engines.shared.helpers.strategy_runtime import (
    load_strategy_settings_view,
)
from core.modules.strategy.engines.shared.performance_profiler import PerformanceProfiler
from core.modules.strategy.services.data import StrategyOutputReaderService
from core.modules.strategy.services.data import StrategyEnumeratorBootstrapService
from core.modules.strategy.services.data.output import (
    SimulationEvent,
    StrategyOutputPathService,
    StrategyOutputVersionService,
)
from core.modules.strategy.services.data.output.event import parse_opportunity_buy_fill
from .data_classes.account import Account, Position
from .data_classes.report import CapitalReport
from .helpers.allocation import AllocationStrategy
from .helpers.core import DateTimeEncoder
from .helpers.fees import FeeCalculator

logger = logging.getLogger(__name__)

if TYPE_CHECKING:
    from core.modules.strategy.engines.shared.data_classes.discovered_strategy import (
        DiscoveredStrategy,
    )


def _downsample_indices(n: int, max_points: int = 80) -> List[int]:
    if n <= 0:
        return []
    if n <= max_points:
        return list(range(n))
    out: List[int] = []
    for k in range(max_points):
        idx = int(round(k * (n - 1) / max(1, max_points - 1)))
        out.append(min(idx, n - 1))
    return sorted(set(out))


def _profit_cv(values: List[float]) -> float:
    vals = [float(v) for v in values if isinstance(v, (int, float))]
    if len(vals) < 2:
        return 0.0
    mean = sum(vals) / len(vals)
    if abs(mean) < 1e-12:
        return 0.0
    var = sum((x - mean) ** 2 for x in vals) / (len(vals) - 1)
    return (var**0.5) / abs(mean)


def _merge_bff_ui_extensions(
    core: Dict[str, Any],
    *,
    trades: List[Dict[str, Any]],
    equity_curve: List[Dict[str, Any]],
    initial_capital: float,
) -> Dict[str, Any]:
    """补充 FED 报告 Tab 所需图表序列与仓位/风险派生指标（与 ``CapitalReport.to_dict()`` 并存）。"""
    out = dict(core)
    ic = float(out.get("initial_capital", initial_capital) or 0.0)
    fe = float(out.get("final_total_equity", out.get("final_equity", 0.0)) or 0.0)
    tr_ratio = float(out.get("total_return", 0.0) or 0.0)
    mdd_ratio = float(out.get("max_drawdown", 0.0) or 0.0)

    calmar = (
        (tr_ratio / mdd_ratio) if mdd_ratio > 1e-12 else 0.0
    )

    points = list(equity_curve or [])
    if not points and ic >= 0:
        fcash = float(out.get("final_cash_balance", fe) or 0.0)
        points = [
            {"date": "期初", "cash_balance": ic, "total_equity": ic, "open_positions": 0},
            {
                "date": "期末",
                "cash_balance": fcash,
                "total_equity": fe,
                "open_positions": 0,
            },
        ]

    labels_full: List[str] = []
    vals_full: List[float] = []
    opens_full: List[float] = []
    cash_full: List[float] = []
    for p in points:
        raw_d = p.get("date", "")
        labels_full.append(str(raw_d)[:16] if raw_d is not None else "")
        vals_full.append(float(p.get("total_equity", 0.0) or 0.0))
        opens_full.append(float(p.get("open_positions", 0) or 0))
        te = float(p.get("total_equity", 0.0) or 1.0)
        cb = float(p.get("cash_balance", 0.0) or 0.0)
        cash_full.append((cb / te * 100.0) if te > 1e-9 else 0.0)

    idxs = _downsample_indices(len(vals_full), 80)
    equity_curve_labels = [labels_full[i] for i in idxs]
    equity_curve_values = [vals_full[i] for i in idxs]

    peak_run = ic
    drawdown_curve_values: List[float] = []
    for v in vals_full:
        peak_run = max(peak_run, v)
        dd_pct = ((peak_run - v) / peak_run * 100.0) if peak_run > 1e-9 else 0.0
        drawdown_curve_values.append(round(dd_pct, 4))
    drawdown_ds = [drawdown_curve_values[i] for i in idxs]

    peak_open = max(opens_full) if opens_full else 0.0
    avg_open = sum(opens_full) / len(opens_full) if opens_full else 0.0
    n_days = len(opens_full)
    full_exp_days = (
        sum(1 for o in opens_full if peak_open > 0 and o >= peak_open - 0.5)
        if peak_open > 0
        else 0
    )
    full_exp_ratio_pct = (full_exp_days / n_days * 100.0) if n_days else 0.0
    avg_cash_pct = sum(cash_full) / len(cash_full) if cash_full else 0.0
    cap_util_pct = max(0.0, min(100.0, 100.0 - avg_cash_pct))

    dd_thresh = 0.5
    max_dd_duration = 0
    run_len = 0
    for dd in drawdown_curve_values:
        if dd > dd_thresh:
            run_len += 1
            max_dd_duration = max(max_dd_duration, run_len)
        else:
            run_len = 0

    sell_trades = [t for t in trades if t.get("side") == "sell"]
    sell_pnls = [float(t.get("pnl", 0.0) or 0.0) for t in sell_trades]
    sell_pnls.sort()
    worst3 = sell_pnls[:3]
    while len(worst3) < 3:
        worst3.append(0.0)

    max_loss_streak = 0
    streak = 0
    for t in sorted(sell_trades, key=lambda x: str(x.get("date", ""))):
        pnl = float(t.get("pnl", 0.0) or 0.0)
        if pnl < 0:
            streak += 1
            max_loss_streak = max(max_loss_streak, streak)
        else:
            streak = 0

    stock_summary = out.get("stock_summary") or {}
    per_stock_profit = [
        float(v.get("total_profit", 0.0) or 0.0)
        for v in stock_summary.values()
        if isinstance(v, dict)
    ]
    positive_sum = sum(x for x in per_stock_profit if x > 0)
    top5_sum = sum(sorted((x for x in per_stock_profit if x > 0), reverse=True)[:5])
    top5_conc_pct = (top5_sum / positive_sum * 100.0) if positive_sum > 1e-9 else 0.0
    stock_cv = _profit_cv(per_stock_profit)

    sell_n = int(out.get("sell_trades", 0) or 0)
    stock_cnt = len(stock_summary) if stock_summary else 0
    avg_trades_ps = sell_n / stock_cnt if stock_cnt else 0.0

    out["equity_curve_labels"] = equity_curve_labels
    out["equity_curve_values"] = equity_curve_values
    out["drawdown_curve_values"] = drawdown_ds
    out["calmar_ratio"] = round(calmar, 4)
    out["average_open_positions"] = round(avg_open, 3)
    out["peak_open_positions"] = int(round(peak_open))
    out["full_exposure_days_ratio_pct"] = round(full_exp_ratio_pct, 2)
    out["average_cash_ratio_pct"] = round(avg_cash_pct, 2)
    out["capital_utilization_ratio_pct"] = round(cap_util_pct, 2)
    out["max_consecutive_losing_sells"] = int(max_loss_streak)
    out["max_drawdown_duration_days"] = int(max_dd_duration)
    out["worst_sell_pnls"] = [round(x, 2) for x in worst3[:3]]
    out["stock_count"] = stock_cnt
    out["average_trades_per_stock"] = round(avg_trades_ps, 4)
    out["top5_profit_concentration_pct"] = round(top5_conc_pct, 2)
    out["stock_profit_coefficient_of_variation"] = round(stock_cv, 4)

    tp = float(out.get("total_profit", 0.0) or 0.0)
    out["profit"] = tp
    out["initialCapital"] = ic
    out["endCapital"] = fe
    ret_pct_out = tr_ratio * 100.0 if abs(tr_ratio) <= 2.0 else tr_ratio
    out["retPct"] = round(ret_pct_out, 4)

    return out


class CapitalAllocationFlowImpl:
    def __init__(self, is_verbose: bool = False) -> None:
        self.is_verbose = is_verbose

    def load_settings(
        self, strategy_name: str, strategy_info: "DiscoveredStrategy | None"
    ):
        return load_strategy_settings_view(strategy_name, strategy_info=strategy_info)

    def parse_config(
        self, base_settings: StrategySettingsView
    ) -> StrategyCapitalSimulatorSettings:
        config = StrategyCapitalSimulatorSettings.from_strategy_root(
            base_settings.to_dict()
        )
        config.apply_defaults()
        return config

    def resolve_source_version(
        self,
        *,
        strategy_name: str,
        base_settings,
        config: StrategyCapitalSimulatorSettings,
        strategy_info: "DiscoveredStrategy | None",
    ) -> Path:
        base_output_version_dir, _ = (
            StrategyEnumeratorBootstrapService.resolve_or_build_enumerator_version(
            strategy_name=strategy_name,
            base_settings=base_settings,
            use_sampling=config.use_sampling,
            base_version=config.base_version,
            strategy_info=strategy_info,
            )
        )
        return base_output_version_dir

    def create_output_version(self, strategy_name: str):
        return StrategyOutputVersionService.create_capital_allocation_version(
            strategy_name
        )

    def create_profiler(self) -> PerformanceProfiler:
        profiler = PerformanceProfiler(stock_id="capital_allocation")
        profiler.start_timer("total")
        return profiler

    def load_event_stream(
        self,
        *,
        strategy_name: str,
        base_output_version_dir: Path,
        config: StrategyCapitalSimulatorSettings,
        base_settings: StrategySettingsView,
        profiler: PerformanceProfiler,
    ) -> List[SimulationEvent]:
        from core.modules.data_manager import DataManager
        from core.modules.strategy.engines.shared.helpers.backtest_date_resolve import (
            resolve_backtest_date_range,
            resolve_latest_completed_trading_date,
        )
        from core.modules.strategy.services.data.output.enumerator_output_service import (
            EnumeratorOutputWriterService,
        )

        profiler.start_timer("load_data")
        stock_ids = EnumeratorOutputWriterService.read_scope_stock_ids(base_output_version_dir)
        data_mgr = DataManager(is_verbose=False)
        period = resolve_backtest_date_range(
            settings_view=base_settings,
            stock_ids=stock_ids,
            latest_completed_trading_date=resolve_latest_completed_trading_date(data_mgr),
            data_manager=data_mgr,
        )
        data_loader = StrategyOutputReaderService(
            strategy_name=strategy_name, cache_enabled=True
        )
        events = data_loader.build_event_stream(
            base_output_version_dir,
            start_date=period.start_date,
            end_date=period.end_date,
        )
        profiler.metrics.time_load_data = profiler.end_timer("load_data")
        profiler.metrics.opportunity_count = len(events)
        if not events:
            logger.warning(
                "资金模拟无事件：枚举目录 %s 下无有效买入（须含 buy_date/buy_price），"
                "请先跑 enum 并确认 sampling.pool 路径相对策略目录名 %s",
                base_output_version_dir,
                strategy_name,
            )
        return events

    def create_execution_state(
        self,
        config: StrategyCapitalSimulatorSettings,
        *,
        market_profile: MarketProfileContext,
        simulation_settings: StrategySimulationSettings,
        backtest_calendar: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        allocation_cfg = config.allocation
        fees_cfg = config.get_fees_config_with_priority()
        account = Account(initial_cash=config.initial_capital, cash=config.initial_capital)
        fee_calculator = FeeCalculator(
            commission_rate=float(fees_cfg.get("commission_rate", 0.00025) or 0.00025),
            min_commission=float(fees_cfg.get("min_commission", 5.0) or 5.0),
            stamp_duty_rate=float(fees_cfg.get("stamp_duty_rate", 0.001) or 0.001),
            transfer_fee_rate=float(fees_cfg.get("transfer_fee_rate", 0.0) or 0.0),
        )
        allocation_strategy = AllocationStrategy(
            mode=allocation_cfg.mode,
            initial_capital=config.initial_capital,
            max_portfolio_size=allocation_cfg.max_portfolio_size,
            market_profile=market_profile.profile,
            lots_per_trade=allocation_cfg.lots_per_trade,
            kelly_fraction=allocation_cfg.kelly_fraction,
            fee_calculator=fee_calculator,
            allow_buy_at_limit_up=simulation_settings.allow_buy_at_limit_up,
            allow_sell_at_limit_down=simulation_settings.allow_sell_at_limit_down,
            skip_trade_when_insufficient=allocation_cfg.skip_trade_when_insufficient,
            skip_investment_when=simulation_settings.skip_investment_when,
            max_participation_rate=simulation_settings.max_participation_rate,
            participation_on_exceed=simulation_settings.participation_on_exceed,
        )
        return {
            "account": account,
            "fee_calculator": fee_calculator,
            "allocation_strategy": allocation_strategy,
            "trades": [],
            "equity_curve": [],
            "current_date": None,
            "completed_opportunities_map": {},
            "tradability_skips": {
                "buy_at_limit_up": 0,
                "sell_at_limit_down": 0,
                "stock_status": 0,
                "buy_participation_skip": 0,
                "buy_participation_clipped": 0,
                "sell_participation_skip": 0,
                "sell_participation_clipped": 0,
            },
            "backtest_calendar": dict(backtest_calendar or {}),
        }

    def replay_events(
        self,
        *,
        events: List[SimulationEvent],
        config: StrategyCapitalSimulatorSettings,
        state: Dict[str, Any],
        profiler: PerformanceProfiler,
        progress_callback: Optional[Callable[[float], None]] = None,
    ) -> None:
        profiler.start_timer("enumerate")
        account = state["account"]
        allocation_strategy = state["allocation_strategy"]
        fee_calculator = state["fee_calculator"]
        n_ev = len(events)
        step_report = max(1, n_ev // 80)
        for i, event in enumerate(events):
            if event.date != state["current_date"]:
                if state["current_date"] is not None and config.output.save_equity_curve:
                    state["equity_curve"].append(
                        {
                            "date": state["current_date"],
                            "cash_balance": account.cash,
                            "total_equity": account.get_equity({}),
                            "open_positions": account.get_portfolio_size(),
                        }
                    )
                state["current_date"] = event.date

            if event.is_trigger():
                trade = self._handle_trigger_event(
                    event,
                    account,
                    allocation_strategy,
                    state["completed_opportunities_map"],
                    tradability_skips=state["tradability_skips"],
                )
                if trade:
                    state["trades"].append(trade)
            elif event.is_target():
                trade = self._handle_target_event(
                    event,
                    account,
                    fee_calculator,
                    allocation_strategy,
                    completed_opportunities_map=state["completed_opportunities_map"],
                    tradability_skips=state["tradability_skips"],
                )
                if trade:
                    state["trades"].append(trade)
            if progress_callback is not None and n_ev:
                if i % step_report == 0 or i == n_ev - 1:
                    frac = (i + 1) / n_ev
                    progress_callback(min(88.0, 15.0 + frac * 73.0))
        profiler.metrics.time_enumerate = profiler.end_timer("enumerate")

    def finalize_equity_curve(
        self,
        *,
        config: StrategyCapitalSimulatorSettings,
        state: Dict[str, Any],
    ) -> None:
        if state["current_date"] is not None and config.output.save_equity_curve:
            account = state["account"]
            state["equity_curve"].append(
                {
                    "date": state["current_date"],
                    "cash_balance": account.cash,
                    "total_equity": account.get_equity({}),
                    "open_positions": account.get_portfolio_size(),
                }
            )

    def build_summary(
        self,
        *,
        account: Account,
        trades: List[Dict[str, Any]],
        equity_curve: List[Dict[str, Any]],
        initial_capital: float,
        events: List[SimulationEvent],
        completed_opportunities_map: Dict[str, Dict[str, Any]],
        tradability_skips: Optional[Dict[str, int]] = None,
    ) -> Dict[str, Any]:
        core = self._calculate_summary(
            account,
            trades,
            equity_curve,
            initial_capital,
            events,
            completed_opportunities_map,
        )
        skips = tradability_skips or {}
        core["skipped_buy_at_limit_up"] = int(skips.get("buy_at_limit_up", 0) or 0)
        core["skipped_sell_at_limit_down"] = int(skips.get("sell_at_limit_down", 0) or 0)
        core["skipped_stock_status"] = int(skips.get("stock_status", 0) or 0)
        core["skipped_buy_participation"] = int(skips.get("buy_participation_skip", 0) or 0)
        core["skipped_sell_participation"] = int(skips.get("sell_participation_skip", 0) or 0)
        core["clipped_buy_participation"] = int(skips.get("buy_participation_clipped", 0) or 0)
        core["clipped_sell_participation"] = int(skips.get("sell_participation_clipped", 0) or 0)
        return _merge_bff_ui_extensions(
            core,
            trades=trades,
            equity_curve=equity_curve,
            initial_capital=float(initial_capital or 0.0),
        )

    def save_outputs(
        self,
        *,
        output_version_dir: Path,
        output_version_id: int,
        base_output_version_dir: Path,
        trades: List[Dict[str, Any]],
        equity_curve: List[Dict[str, Any]],
        summary: Dict[str, Any],
        config: StrategyCapitalSimulatorSettings,
        settings_snapshot: Dict[str, Any],
        simulation_effective: Optional[Dict[str, Any]] = None,
    ) -> None:
        self._save_results(
            output_version_dir,
            output_version_id,
            base_output_version_dir,
            trades,
            equity_curve,
            summary,
            config,
            settings_snapshot,
            simulation_effective,
        )

    def save_performance_report(
        self, *, output_version_dir: Path, profiler: PerformanceProfiler
    ) -> None:
        try:
            perf_path = (
                StrategyOutputPathService(output_version_dir=output_version_dir).ensure_root()
                / "0_performance_report.json"
            )
            with perf_path.open("w", encoding="utf-8") as file:
                json.dump(
                    profiler.finalize().to_dict(),
                    file,
                    indent=2,
                    ensure_ascii=False,
                    cls=DateTimeEncoder,
                )
        except Exception:
            pass

    def run_analyzer_hook(
        self, *, strategy_name: str, output_version_dir: Path, raw_settings: Dict[str, Any]
    ) -> None:
        try:
            Analyzer.run_for_simulator(
                strategy_name=strategy_name,
                sim_type="capital_allocation",
                output_version_dir=output_version_dir,
                raw_settings=raw_settings,
            )
        except Exception:
            pass

    def _handle_trigger_event(
        self,
        event: SimulationEvent,
        account: Account,
        allocation_strategy: AllocationStrategy,
        completed_opportunities_map: Dict[str, Dict[str, Any]],
        *,
        tradability_skips: Optional[Dict[str, int]] = None,
    ) -> Optional[Dict[str, Any]]:
        stock_id = event.stock_id
        opportunity = event.opportunity or {}
        opp_id = str(opportunity.get("opportunity_id", "")).strip()
        buy_fill = parse_opportunity_buy_fill(opportunity)
        if not stock_id or not opp_id or buy_fill is None:
            if opp_id and stock_id and buy_fill is None:
                logger.warning(
                    "跳过买入：机会缺少有效 buy_date/buy_price stock=%s opp=%s",
                    stock_id,
                    opp_id,
                )
            return None
        buy_event_date, buy_price = buy_fill
        skip_status = should_skip_investment(
            opportunity,
            allocation_strategy.skip_investment_when,
        )
        if skip_status:
            if tradability_skips is not None:
                tradability_skips["stock_status"] = (
                    int(tradability_skips.get("stock_status", 0) or 0) + 1
                )
            return None
        if should_skip_buy(
            opportunity,
            allocation_strategy.market_profile,
            stock_id,
            buy_price,
            allow_at_limit=allocation_strategy.allow_buy_at_limit_up,
        ):
            if tradability_skips is not None:
                tradability_skips["buy_at_limit_up"] = (
                    int(tradability_skips.get("buy_at_limit_up", 0) or 0) + 1
                )
            return None
        if (
            account.has_position(stock_id)
            or account.get_portfolio_size() >= allocation_strategy.max_portfolio_size
        ):
            return None
        win_rate = (
            self._calculate_win_rate(completed_opportunities_map)
            if allocation_strategy.mode == "kelly"
            else None
        )
        buy_shares = allocation_strategy.calculate_shares_to_buy(
            account, buy_price, stock_id, win_rate
        )
        if buy_shares == 0:
            return None
        buy_shares, part_tag = apply_participation_to_shares(
            buy_shares,
            bar_volume=row_buy_bar_volume(opportunity),
            max_participation_rate=allocation_strategy.max_participation_rate,
            on_exceed=allocation_strategy.participation_on_exceed,
            floor_shares_fn=lambda n, _sid: allocation_strategy._floor_buy_shares(
                n, stock_id
            ),
            stock_id=stock_id,
        )
        if part_tag == "participation_skip" or part_tag == "participation_clip_zero":
            if tradability_skips is not None:
                tradability_skips["buy_participation_skip"] = (
                    int(tradability_skips.get("buy_participation_skip", 0) or 0) + 1
                )
            return None
        if part_tag == "participation_clipped" and tradability_skips is not None:
            tradability_skips["buy_participation_clipped"] = (
                int(tradability_skips.get("buy_participation_clipped", 0) or 0) + 1
            )
        if buy_shares == 0:
            return None
        gross_amount = buy_shares * buy_price
        fees = allocation_strategy.fee_calculator.calculate_fees(gross_amount, "buy")
        total_cost = gross_amount + fees
        if account.cash < total_cost:
            return None
        account.cash -= total_cost
        position = account.positions.get(stock_id) or Position(stock_id=stock_id)
        account.positions[stock_id] = position
        position.shares = buy_shares
        position.avg_cost = total_cost / buy_shares
        position.current_opportunity_id = opp_id
        return {
            "date": buy_event_date,
            "stock_id": stock_id,
            "opportunity_id": opp_id,
            "side": "buy",
            "shares": buy_shares,
            "price": buy_price,
            "amount": gross_amount,
            "fees": fees,
            "total_cost": total_cost,
            "cash_after": account.cash,
            "equity_after": account.get_equity({stock_id: buy_price}),
        }

    def _handle_target_event(
        self,
        event: SimulationEvent,
        account: Account,
        fee_calculator: FeeCalculator,
        allocation_strategy: AllocationStrategy,
        *,
        completed_opportunities_map: Dict[str, Dict[str, Any]],
        tradability_skips: Optional[Dict[str, int]] = None,
    ) -> Optional[Dict[str, Any]]:
        stock_id = event.stock_id
        target = event.target or {}
        opp_id = str(target.get("opportunity_id", "")).strip()
        raw_sell_price = (
            target.get("sell_price", 0.0)
            or target.get("price", 0.0)
            or target.get("target_price", 0.0)
        )
        sell_price = float(raw_sell_price or 0.0)
        if not stock_id or not opp_id or sell_price <= 0:
            return None
        if should_skip_sell(
            target,
            allocation_strategy.market_profile,
            stock_id,
            sell_price,
            allow_at_limit=allocation_strategy.allow_sell_at_limit_down,
        ):
            if tradability_skips is not None:
                tradability_skips["sell_at_limit_down"] = (
                    int(tradability_skips.get("sell_at_limit_down", 0) or 0) + 1
                )
            return None
        position = account.get_position(stock_id)
        if (
            not position
            or position.shares == 0
            or position.current_opportunity_id != opp_id
        ):
            return None
        raw_sell_ratio = target.get("sell_ratio", 0.0)
        try:
            sell_ratio = float(
                raw_sell_ratio if raw_sell_ratio not in (None, "", 0) else 1.0
            )
        except (TypeError, ValueError):
            sell_ratio = 1.0
        sell_shares = int(position.shares * sell_ratio)
        if sell_shares == 0:
            return None
        sell_shares, part_tag = apply_participation_to_shares(
            sell_shares,
            bar_volume=row_sell_bar_volume(target),
            max_participation_rate=allocation_strategy.max_participation_rate,
            on_exceed=allocation_strategy.participation_on_exceed,
            floor_shares_fn=lambda n, _sid: allocation_strategy._floor_buy_shares(
                n, stock_id
            ),
            stock_id=stock_id,
        )
        if part_tag == "participation_skip" or part_tag == "participation_clip_zero":
            if tradability_skips is not None:
                tradability_skips["sell_participation_skip"] = (
                    int(tradability_skips.get("sell_participation_skip", 0) or 0) + 1
                )
            return None
        if part_tag == "participation_clipped" and tradability_skips is not None:
            tradability_skips["sell_participation_clipped"] = (
                int(tradability_skips.get("sell_participation_clipped", 0) or 0) + 1
            )
        if sell_shares == 0:
            return None
        gross_amount = sell_shares * sell_price
        fees = fee_calculator.calculate_fees(gross_amount, "sell")
        net_proceeds = gross_amount - fees
        cost = sell_shares * position.avg_cost
        pnl = net_proceeds - cost
        account.cash += net_proceeds
        position.shares -= sell_shares
        position.realized_pnl += pnl
        if position.shares == 0:
            # 须在清空 ``current_opportunity_id`` 之前记入已完成；否则下游无法用 id 对齐。
            opportunity = dict(event.opportunity or {})
            if not str(opportunity.get("opportunity_id") or "").strip():
                opportunity["opportunity_id"] = opp_id
            if opp_id not in completed_opportunities_map:
                completed_opportunities_map[opp_id] = opportunity
            position.current_opportunity_id = None
        return {
            "date": event.date,
            "stock_id": stock_id,
            "opportunity_id": opp_id,
            "side": "sell",
            "shares": sell_shares,
            "price": sell_price,
            "amount": gross_amount,
            "fees": fees,
            "net_proceeds": net_proceeds,
            "pnl": pnl,
            "cash_after": account.cash,
            "equity_after": account.get_equity({stock_id: sell_price}),
        }

    def _calculate_win_rate(
        self, completed_opportunities_map: Dict[str, Dict[str, Any]]
    ) -> float:
        if not completed_opportunities_map:
            return 0.5
        opportunities = list(completed_opportunities_map.values())
        win_count = sum(
            1 for opp in opportunities if float(opp.get("roi", 0) or 0) > 0
        )
        return win_count / len(opportunities) if opportunities else 0.5

    def _calculate_summary(
        self,
        account: Account,
        trades: List[Dict[str, Any]],
        equity_curve: List[Dict[str, Any]],
        initial_capital: float,
        events: List[SimulationEvent],
        completed_opportunities_map: Dict[str, Dict[str, Any]],
    ) -> Dict[str, Any]:
        final_equity = (
            equity_curve[-1].get("total_equity", account.cash)
            if equity_curve
            else account.cash
        )
        total_return = (
            (final_equity - initial_capital) / initial_capital
            if initial_capital > 0
            else 0.0
        )
        max_drawdown = 0.0
        if equity_curve:
            peak = initial_capital
            for point in equity_curve:
                equity = point.get("total_equity", initial_capital)
                if equity > peak:
                    peak = equity
                max_drawdown = max(
                    max_drawdown, (peak - equity) / peak if peak > 0 else 0.0
                )
        buy_trades = [t for t in trades if t.get("side") == "buy"]
        sell_trades = [t for t in trades if t.get("side") == "sell"]
        win_trades = [t for t in sell_trades if t.get("pnl", 0) > 0]
        loss_trades = [t for t in sell_trades if t.get("pnl", 0) < 0]
        stock_summary = defaultdict(
            lambda: {
                "trade_count": 0,
                "total_profit": 0.0,
                "win_trades": 0,
                "loss_trades": 0,
            }
        )
        for trade in sell_trades:
            stock_id = trade.get("stock_id", "")
            stock_summary[stock_id]["trade_count"] += 1
            stock_summary[stock_id]["total_profit"] += trade.get("pnl", 0.0)
            if trade.get("pnl", 0) > 0:
                stock_summary[stock_id]["win_trades"] += 1
            else:
                stock_summary[stock_id]["loss_trades"] += 1
        trigger_ids = {
            str(e.opportunity_id or "").strip()
            for e in events
            if e.is_trigger() and str(e.opportunity_id or "").strip()
        }
        completed_ids = {
            str(opp_id).strip()
            for opp_id in completed_opportunities_map.keys()
            if str(opp_id).strip()
        }
        total_opportunities = len(trigger_ids)
        completed_opportunities = len(trigger_ids & completed_ids)
        unfinished_opportunities = max(total_opportunities - completed_opportunities, 0)
        raw_summary = {
            "initial_capital": initial_capital,
            "final_cash_balance": account.cash,
            "final_total_equity": final_equity,
            "final_equity": final_equity,
            "total_return": total_return,
            "max_drawdown": max_drawdown,
            "total_trades": len(trades),
            "buy_trades": len(buy_trades),
            "sell_trades": len(sell_trades),
            "win_trades": len(win_trades),
            "loss_trades": len(loss_trades),
            "win_rate": len(win_trades) / len(sell_trades) if sell_trades else 0.0,
            "total_profit": sum(t.get("pnl", 0.0) for t in sell_trades),
            "total_opportunities": total_opportunities,
            "completed_opportunities": completed_opportunities,
            "unfinished_opportunities": unfinished_opportunities,
            "completion_rate": (
                (completed_opportunities / total_opportunities)
                if total_opportunities > 0
                else 0.0
            ),
            "stock_summary": dict(stock_summary),
        }
        return CapitalReport.from_dict(raw_summary).to_dict()

    def _save_results(
        self,
        output_version_dir: Path,
        output_version_id: int,
        base_output_version_dir: Path,
        trades: List[Dict[str, Any]],
        equity_curve: List[Dict[str, Any]],
        summary: Dict[str, Any],
        config: StrategyCapitalSimulatorSettings,
        settings_snapshot: Dict[str, Any],
        simulation_effective: Optional[Dict[str, Any]] = None,
    ) -> None:
        from core.utils.io.csv_io import write_dicts_to_csv

        path_mgr = StrategyOutputPathService(output_version_dir=output_version_dir)
        if config.output.save_trades and trades:
            trades_path = path_mgr.trades_path()
            with trades_path.open("w", encoding="utf-8") as f:
                json.dump(trades, f, indent=2, ensure_ascii=False, cls=DateTimeEncoder)
            write_dicts_to_csv(trades_path.with_suffix(".csv"), trades)
        if config.output.save_equity_curve and equity_curve:
            equity_path = path_mgr.equity_timeseries_path()
            with equity_path.open("w", encoding="utf-8") as f:
                json.dump(
                    equity_curve, f, indent=2, ensure_ascii=False, cls=DateTimeEncoder
                )
            write_dicts_to_csv(equity_path.with_suffix(".csv"), equity_curve)
        with path_mgr.strategy_summary_path().open("w", encoding="utf-8") as f:
            json.dump(summary, f, indent=2, ensure_ascii=False, cls=DateTimeEncoder)
        metadata = {
            "output_version_run": {
                "output_version_id": output_version_id,
                "output_version_dir": output_version_dir.name,
                "run_label": f"{output_version_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}",
            },
            "output_version": {
                "enumerator_output_dir": base_output_version_dir.name,
                "output_root": str(base_output_version_dir.parent.name),
            },
            "config": config.to_dict(),
            "settings_snapshot": settings_snapshot,
            "created_at": datetime.now().isoformat(),
        }
        if simulation_effective:
            metadata["simulation"] = simulation_effective
        with path_mgr.metadata_path().open("w", encoding="utf-8") as f:
            json.dump(metadata, f, indent=2, ensure_ascii=False, cls=DateTimeEncoder)

__all__ = ["CapitalAllocationFlowImpl"]
