external/Quant.ai/backend/app/simulator.py

# backend/app/simulator.py

import pandas as pd
import numpy as np
import datetime
from app.config import INITIAL_CASH, FORCE_LIQUIDATION_TIME, SOFT_DRAWDOWN_LIMIT, HARD_DRAWDOWN_LIMIT, MAX_CONSECUTIVE_LOSSES, FORCE_LIQUIDATION_OPEN_FOCUS
from app.trading_engine import Portfolio
from app.strategy import evaluate_market_state

def run_backtest_sim(df, ticker, strategy_params, risk_params, is_intraday=True):
    """
    通用回测执行引擎，支持日内 (is_intraday=True) 和日线 (is_intraday=False) 级别回测。
    增强输出：regime_breakdown, drawdown_curve, Sharpe, Calmar, CAGR, Profit Factor
    """
    df = df.copy()
    if is_intraday:
        # Precompute ORB (Opening Range Breakout) High/Low for each day
        # Range is defined between 9:30 and 9:35 (first 5 minutes of regular trading hours)
        df['Time'] = df.index.time
        df['Date'] = df.index.date
        
        import datetime as dt_mod
        open_mask = (df['Time'] >= dt_mod.time(9, 30)) & (df['Time'] <= dt_mod.time(9, 35))
        opening_df = df[open_mask]
        
        if not opening_df.empty:
            orb_high = opening_df.groupby('Date')['High'].max().to_dict()
            orb_low = opening_df.groupby('Date')['Low'].min().to_dict()
            df['ORB_High'] = df['Date'].map(orb_high).fillna(0.0)
            df['ORB_Low'] = df['Date'].map(orb_low).fillna(0.0)
        else:
            df['ORB_High'] = 0.0
            df['ORB_Low'] = 0.0

    portfolio = Portfolio(
        initial_cash=INITIAL_CASH,
        slippage_rate=risk_params.get("slippage_rate", 0.0003),
        commission_per_share=risk_params.get("commission_per_share", 0.005),
        min_commission_per_order=risk_params.get("min_commission_per_order", 1.0)
    )
    
    # 提取账户风控限制
    soft_dd = risk_params.get("soft_dd", SOFT_DRAWDOWN_LIMIT)
    hard_dd = risk_params.get("hard_dd", HARD_DRAWDOWN_LIMIT)
    max_consecutive_losses = risk_params.get("max_consecutive_losses", MAX_CONSECUTIVE_LOSSES)
    
    # 提取开盘突击模式参数，默认是 True
    market_open_focus = strategy_params.get("market_open_focus", True)
    
    equity_curve = []
    
    # 记录每笔交易所处的 regime（用于 regime breakdown）
    trade_regime_map = {}  # buy_timestamp -> regime
    
    # 我们从第2行开始，因为策略需要 prev_row 进行对比
    for i in range(1, len(df)):
        row = df.iloc[i]
        prev_row = df.iloc[i-1]
        timestamp = df.index[i]
        time_str = timestamp.strftime("%H:%M")
        
        close_price = float(row['Close'])
        current_prices = {ticker: close_price}
        
        shares = portfolio.get_position_shares(ticker)
        avg_cost = portfolio.get_position_avg_cost(ticker)
        
        # 1. 账户级风控指标更新 (含最高权益及回撤计算)
        equity = portfolio.get_equity(current_prices)
        portfolio.peak_equity = max(portfolio.peak_equity, equity)
        drawdown_pct = (portfolio.peak_equity - equity) / portfolio.peak_equity if portfolio.peak_equity > 0 else 0.0
        
        # 统计最近 SELL 操作的 realized_pnl
        sell_trades = [t for t in portfolio.ledger if t['action'] == 'SELL']
        consecutive_losses = 0
        for t in reversed(sell_trades):
            if t.get('realized_pnl', 0.0) < 0:
                consecutive_losses += 1
            else:
                break
        portfolio.consecutive_losses = consecutive_losses
        
        # 2. 判定硬/软熔断并更新 risk_multiplier
        if drawdown_pct >= hard_dd:
            portfolio.risk_multiplier = 0.0
        elif drawdown_pct >= soft_dd or consecutive_losses >= max_consecutive_losses:
            portfolio.risk_multiplier = 0.5
        else:
            portfolio.risk_multiplier = 1.0
            
        # 3. 更新追踪止损的最高价
        if shares > 0:
            portfolio.update_highest_price(ticker, close_price)
            
        highest_price = portfolio.get_position_highest_price(ticker)
        
        # 记录资产曲线数据
        equity_curve.append({
            "time": int(timestamp.timestamp()),
            "value": round(equity, 2)
        })
        
        # 4. 日内清仓检测 (仅限 1分钟、5分钟等日内级别)
        liq_time = FORCE_LIQUIDATION_OPEN_FOCUS if market_open_focus else FORCE_LIQUIDATION_TIME
        if is_intraday and time_str == liq_time and shares > 0:
            portfolio.sell(timestamp, ticker, close_price, shares)
            continue
            
        # 5. 执行常规策略评估
        is_trading_window = True
        if is_intraday:
            if market_open_focus:
                # 开盘突击：只在 9:35 - 10:15 之间开新仓
                import datetime as dt_mod
                is_trading_window = dt_mod.time(9, 35) <= timestamp.time() <= dt_mod.time(10, 15)
            else:
                # 日内交易时间窗口：上午 9:35 到 下午 15:54
                is_trading_window = datetime.time(9, 35) <= timestamp.time() < datetime.time(15, 54)
                
        # 如果我们已经持有仓位，为了检查止损止盈等，任何时间（直到强制平仓前）都是交易窗口
        if shares > 0:
            is_trading_window = True
        if is_trading_window:
            current_regime = row.get('Regime', 'range_bound')
            
            action, explanation = evaluate_market_state(
                row, prev_row, shares, avg_cost, ticker, highest_price, strategy_params
            )
            
            if action == "BUY" and shares == 0:
                # 仓位计算
                if risk_params.get("position_sizing_mode", "atr") == "atr":
                    shares_to_buy = portfolio.calculate_position_size(
                        ticker, close_price, row['ATR'],
                        risk_pct=risk_params.get("risk_per_trade_pct", 0.01),
                        atr_multiplier=strategy_params.get("trailing_stop_atr_mult", 2.0),
                        max_size_pct=risk_params.get("max_position_size_pct", 0.50)
                    )
                else:
                    target_allocation = equity * risk_params.get("max_position_size_pct", 0.50)
                    shares_to_buy = int(target_allocation / close_price)
                    
                if shares_to_buy > 0:
                    portfolio.buy(timestamp, ticker, close_price, shares_to_buy)
                    # 记录买入时的 regime
                    trade_regime_map[str(timestamp)] = current_regime
                    
            elif action == "SELL" and shares > 0:
                portfolio.sell(timestamp, ticker, close_price, shares)
                
    # 6. 计算回测指标
    total_trades = len(portfolio.ledger)
    final_equity = portfolio.get_equity({ticker: df.iloc[-1]['Close']})
    net_pnl = final_equity - INITIAL_CASH
    pnl_pct = (net_pnl / INITIAL_CASH) * 100
    
    sell_trades = [t for t in portfolio.ledger if t['action'] == 'SELL']
    winning_trades = [t for t in sell_trades if t.get('realized_pnl', 0.0) > 0]
    losing_trades = [t for t in sell_trades if t.get('realized_pnl', 0.0) < 0]
    win_rate = (len(winning_trades) / len(sell_trades)) * 100 if sell_trades else 0.0
    
    total_commission = sum(t['commission'] for t in portfolio.ledger)
    
    # 统计最大回撤与 drawdown curve
    eq_vals = [e["value"] for e in equity_curve]
    max_drawdown = 0.0
    drawdown_curve = []
    if eq_vals:
        peaks = np.maximum.accumulate(eq_vals)
        drawdowns = (peaks - eq_vals) / peaks
        max_drawdown = float(np.max(drawdowns))
        for idx_ec, ec in enumerate(equity_curve):
            drawdown_curve.append({
                "time": ec["time"],
                "value": round(float(drawdowns[idx_ec]) * -100, 2)  # 负百分比
            })
    
    # 7. 高级指标计算
    # Profit Factor = 总盈利 / 总亏损
    gross_profit = sum(t.get('realized_pnl', 0.0) for t in winning_trades)
    gross_loss = abs(sum(t.get('realized_pnl', 0.0) for t in losing_trades))
    profit_factor = round(gross_profit / max(gross_loss, 1e-8), 2)
    
    # 计算回测跨度的交易日数
    if len(df) >= 2:
        start_date = df.index[0]
        end_date = df.index[-1]
        total_days = (end_date - start_date).days
        total_years = total_days / 365.25 if total_days > 0 else 1.0 / 365.25
    else:
        total_years = 1.0 / 365.25
    
    # CAGR = (Final/Initial)^(1/years) - 1
    if final_equity > 0 and INITIAL_CASH > 0 and total_years > 0:
        cagr = (final_equity / INITIAL_CASH) ** (1.0 / max(total_years, 0.01)) - 1.0
    else:
        cagr = 0.0
    
    # Sharpe Ratio (基于日/bar收益序列，假设 risk-free = 0)
    if len(eq_vals) > 1:
        returns = np.diff(eq_vals) / np.array(eq_vals[:-1])
        if np.std(returns) > 0:
            # 年化 Sharpe（按交易日数年化）
            bars_per_year = len(eq_vals) / max(total_years, 0.01)
            sharpe = (np.mean(returns) / np.std(returns)) * np.sqrt(bars_per_year)
        else:
            sharpe = 0.0
    else:
        sharpe = 0.0
    
    # Calmar Ratio = CAGR / Max Drawdown
    calmar = round(cagr / max(max_drawdown, 1e-8), 2) if max_drawdown > 0.001 else 0.0
    
    # 8. Regime Breakdown — 统计不同市场状态下交易表现
    regime_breakdown = _compute_regime_breakdown(portfolio.ledger, trade_regime_map)
    
    # 9. Regime 分布统计 (每个 regime 的 bar 数量占比)
    regime_distribution = {}
    if 'Regime' in df.columns:
        regime_counts = df['Regime'].value_counts()
        total_bars = len(df)
        for regime_name, count in regime_counts.items():
            regime_distribution[regime_name] = round(count / total_bars * 100, 1)
    
    return {
        "final_equity": round(final_equity, 2),
        "net_pnl": round(net_pnl, 2),
        "pnl_pct": round(pnl_pct, 2),
        "total_trades": total_trades,
        "round_trips": len(sell_trades),
        "win_rate": round(win_rate, 2),
        "commission": round(total_commission, 2),
        "max_drawdown": round(max_drawdown, 4),
        # 新增高级指标
        "sharpe": round(sharpe, 2),
        "calmar": calmar,
        "cagr": round(cagr * 100, 2),  # 百分比
        "profit_factor": profit_factor,
        "gross_profit": round(gross_profit, 2),
        "gross_loss": round(gross_loss, 2),
        # 数据序列
        "ledger": portfolio.ledger,
        "equity_curve": equity_curve,
        "drawdown_curve": drawdown_curve,
        # Regime 统计
        "regime_breakdown": regime_breakdown,
        "regime_distribution": regime_distribution,
    }


def _compute_regime_breakdown(ledger, trade_regime_map):
    """
    按 regime 汇总交易表现：pnl, win_rate, trade_count
    """
    regime_stats = {}
    
    # 将 BUY/SELL 配对，并查找每对交易的 regime
    buy_trades = [t for t in ledger if t['action'] == 'BUY']
    sell_trades = [t for t in ledger if t['action'] == 'SELL']
    
    for i, sell in enumerate(sell_trades):
        # 每个 sell 与对应的 buy 配对
        if i < len(buy_trades):
            buy_ts = buy_trades[i]['timestamp']
            regime = trade_regime_map.get(buy_ts, 'unknown')
        else:
            regime = 'unknown'
        
        pnl = sell.get('realized_pnl', 0.0)
        
        if regime not in regime_stats:
            regime_stats[regime] = {
                'total_pnl': 0.0,
                'trade_count': 0,
                'wins': 0,
                'losses': 0,
                'total_commission': 0.0
            }
        
        regime_stats[regime]['total_pnl'] += pnl
        regime_stats[regime]['trade_count'] += 1
        regime_stats[regime]['total_commission'] += sell.get('commission', 0.0)
        if pnl > 0:
            regime_stats[regime]['wins'] += 1
        else:
            regime_stats[regime]['losses'] += 1
    
    # 转为前端友好格式
    result = []
    for regime, stats in regime_stats.items():
        win_rate = (stats['wins'] / stats['trade_count'] * 100) if stats['trade_count'] > 0 else 0.0
        result.append({
            "regime": regime,
            "total_pnl": round(stats['total_pnl'], 2),
            "trade_count": stats['trade_count'],
            "win_rate": round(win_rate, 1),
            "wins": stats['wins'],
            "losses": stats['losses'],
            "commission": round(stats['total_commission'], 2)
        })
    
    # 按 trade_count 降序
    result.sort(key=lambda x: x['trade_count'], reverse=True)
    return result