"""
RaptorBT + Mt5Bridge 集成测试

从 Mt5Bridge API 拉取真实 K 线数据，用 RaptorBT 跑回测。

运行前确保：
  pip install raptorbt requests numpy pandas

运行：
  python test_with_mt5.py
"""

import os
import requests
import numpy as np
import pandas as pd
import raptorbt
from datetime import datetime, timedelta, timezone

BRIDGE = "http://61.164.252.86:13485"
API_KEY = "UiHMqtaYLZzwBdcuS4RFmEGhgDO8N2eI"

SYMBOL = "XAUUSD"
TIMEFRAME = "TIMEFRAME_H1"
BARS = 500


def api_get(path, params=None):
    resp = requests.get(
        f"{BRIDGE}{path}",
        params=params,
        headers={"X-API-Key": API_KEY},
        timeout=15,
    )
    resp.raise_for_status()
    return resp.json()


def check_health():
    print("=" * 60)
    print("1. 健康检查")
    print("=" * 60)
    data = api_get("/health")
    print(f"  状态: {data.get('status')}")
    print(f"  MT5 连接: {data.get('mt5_connected')}")
    if not data.get("mt5_connected"):
        print("  ⚠️ MT5 未连接，后续可能失败")
    return data


def fetch_account():
    print("\n" + "=" * 60)
    print("2. 账户信息")
    print("=" * 60)
    data = api_get("/account")["data"][0]
    print(f"  余额: {data['balance']:.2f} {data['currency']}")
    print(f"  净值: {data['equity']:.2f}")
    print(f"  浮动盈亏: {data['profit']:.2f}")
    print(f"  杠杆: 1:{data['leverage']}")
    return data


def fetch_klines():
    print("\n" + "=" * 60)
    print(f"3. 拉取 K 线数据 ({SYMBOL}, {TIMEFRAME}, 最近 {BARS} 根)")
    print("=" * 60)

    date_to = datetime.now(timezone.utc).strftime("%Y-%m-%d")
    date_from = (datetime.now(timezone.utc) - timedelta(days=BARS // 24 + 30)).strftime("%Y-%m-%d")

    data = api_get("/rates/from-date", params={
        "symbol": SYMBOL,
        "timeframe": TIMEFRAME,
        "date_from": date_from,
        "date_to": date_to,
    })

    rows = data.get("data", [])
    if not rows:
        print("  ⚠️ 没有拉到数据，尝试 /rates/from-pos ...")
        data = api_get("/rates/from-pos", params={
            "symbol": SYMBOL,
            "timeframe": TIMEFRAME,
            "start_pos": 0,
            "count": BARS,
        })
        rows = data.get("data", [])

    if not rows:
        raise RuntimeError("两种方式都没拉到 K 线，检查品种名或 MT5 连接")

    df = pd.DataFrame(rows)
    df["time"] = pd.to_datetime(df["time"])
    df = df.sort_values("time").reset_index(drop=True)

    print(f"  拉到 {len(df)} 根 K 线")
    print(f"  时间范围: {df['time'].iloc[0]} ~ {df['time'].iloc[-1]}")
    print(f"  Close 范围: {df['close'].min():.2f} ~ {df['close'].max():.2f}")
    return df


def run_backtest(df):
    print("\n" + "=" * 60)
    print("4. RaptorBT 回测 (SMA 交叉策略)")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    timestamps = df["time"].values.astype("int64")

    sma_fast = raptorbt.sma(close, period=10)
    sma_slow = raptorbt.sma(close, period=20)

    entries = (sma_fast > sma_slow) & np.roll(sma_fast <= sma_slow, 1)
    exits = (sma_fast < sma_slow) & np.roll(sma_fast >= sma_slow, 1)
    entries[:20] = False
    exits[:20] = False

    entries = entries.astype(bool)
    exits = exits.astype(bool)

    n_entries = int(entries.sum())
    n_exits = int(exits.sum())
    print(f"  SMA(10)/SMA(20) 交叉信号: {n_entries} 次入场, {n_exits} 次出场")

    config = raptorbt.PyBacktestConfig(
        initial_capital=100000.0,
        fees=0.001,
        slippage=0.0005,
    )
    config.set_fixed_stop(0.02)
    config.set_fixed_target(0.04)

    result = raptorbt.run_single_backtest(
        timestamps=timestamps,
        open=open_,
        high=high,
        low=low,
        close=close,
        volume=volume,
        entries=entries,
        exits=exits,
        direction=1,
        weight=1.0,
        symbol=SYMBOL,
        config=config,
    )

    m = result.metrics
    print(f"\n  {'─' * 40}")
    print(f"  回测结果")
    print(f"  {'─' * 40}")
    print(f"  总收益率:     {m.total_return_pct:>10.2f} %")
    print(f"  夏普比率:     {m.sharpe_ratio:>10.2f}")
    print(f"  索提诺比率:   {m.sortino_ratio:>10.2f}")
    print(f"  卡玛比率:     {m.calmar_ratio:>10.2f}")
    print(f"  最大回撤:     {m.max_drawdown_pct:>10.2f} %")
    print(f"  总交易数:     {m.total_trades:>10d}")
    print(f"  胜率:         {m.win_rate_pct:>10.1f} %")
    print(f"  盈利因子:     {m.profit_factor:>10.2f}")
    print(f"  平均交易收益: {m.avg_trade_return_pct:>10.2f} %")
    print(f"  平均盈利:     {m.avg_win_pct:>10.2f} %")
    print(f"  平均亏损:     {m.avg_loss_pct:>10.2f} %")
    print(f"  最佳交易:     {m.best_trade_pct:>10.2f} %")
    print(f"  最差交易:     {m.worst_trade_pct:>10.2f} %")
    print(f"  期望值:       {m.expectancy:>10.2f}")
    print(f"  SQN:          {m.sqn:>10.2f}")
    print(f"  总手续费:     {m.total_fees_paid:>10.2f}")
    print(f"  市场暴露:     {m.exposure_pct:>10.1f} %")

    trades = result.trades()
    if trades:
        print(f"\n  最近 5 笔交易:")
        print(f"  {'ID':>6} {'方向':>4} {'入场价':>12} {'出场价':>12} {'收益%':>10} {'出场原因':>12}")
        for t in trades[-5:]:
            direction = "多" if t.direction == 1 else "空"
            print(f"  {t.id:>6} {direction:>4} {t.entry_price:>12.5f} {t.exit_price:>12.5f} {t.return_pct:>10.2f} {t.exit_reason:>12}")

    return result


def run_multi_indicator_backtest(df):
    print("\n" + "=" * 60)
    print("5. 多策略回测 (SMA交叉 + RSI均值回归)")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    timestamps = df["time"].values.astype("int64")

    sma_fast = raptorbt.sma(close, period=10)
    sma_slow = raptorbt.sma(close, period=20)
    rsi = raptorbt.rsi(close, period=14)

    entries_sma = ((sma_fast > sma_slow) & np.roll(sma_fast <= sma_slow, 1)).astype(bool)
    exits_sma = ((sma_fast < sma_slow) & np.roll(sma_fast >= sma_slow, 1)).astype(bool)

    entries_rsi = (rsi < 30).astype(bool)
    exits_rsi = (rsi > 70).astype(bool)

    entries_sma[:20] = False
    exits_sma[:20] = False
    entries_rsi[:14] = False
    exits_rsi[:14] = False

    strategies = [
        (entries_sma, exits_sma, 1, 0.6, "SMA_Cross"),
        (entries_rsi, exits_rsi, 1, 0.4, "RSI_MeanRev"),
    ]

    config = raptorbt.PyBacktestConfig(
        initial_capital=100000.0,
        fees=0.001,
        slippage=0.0005,
    )
    config.set_trailing_stop(0.03)

    result = raptorbt.run_multi_backtest(
        timestamps=timestamps,
        open=open_,
        high=high,
        low=low,
        close=close,
        volume=volume,
        strategies=strategies,
        config=config,
        combine_mode="any",
    )

    m = result.metrics
    print(f"  总收益率: {m.total_return_pct:.2f}%  夏普: {m.sharpe_ratio:.2f}  "
          f"最大回撤: {m.max_drawdown_pct:.2f}%  交易数: {m.total_trades}  胜率: {m.win_rate_pct:.1f}%")
    return result


def show_indicators(df):
    print("\n" + "=" * 60)
    print("6. 内置指标演示")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)

    indicators = {
        "SMA(20)": raptorbt.sma(close, period=20),
        "EMA(20)": raptorbt.ema(close, period=20),
        "RSI(14)": raptorbt.rsi(close, period=14),
        "ATR(14)": raptorbt.atr(high, low, close, period=14),
        "ADX(14)": raptorbt.adx(high, low, close, period=14),
        "VWAP": raptorbt.vwap(high, low, close, volume),
    }

    macd_line, signal_line, hist = raptorbt.macd(close, 12, 26, 9)
    indicators["MACD_Line"] = macd_line
    indicators["MACD_Signal"] = signal_line

    stoch_k, stoch_d = raptorbt.stochastic(high, low, close, k_period=14, d_period=3)
    indicators["Stoch_K"] = stoch_k
    indicators["Stoch_D"] = stoch_d

    upper, middle, lower = raptorbt.bollinger_bands(close, period=20, std_dev=2.0)
    indicators["BB_Upper"] = upper
    indicators["BB_Lower"] = lower

    supertrend_val, supertrend_dir = raptorbt.supertrend(high, low, close, period=10, multiplier=3.0)
    indicators["Supertrend"] = supertrend_val

    indicators["Rolling_Min(20)"] = raptorbt.rolling_min(low, period=20)
    indicators["Rolling_Max(20)"] = raptorbt.rolling_max(high, period=20)

    last_idx = -1
    print(f"\n  最新一根 K 线的指标值:")
    print(f"  {'指标':<20} {'值':>15}")
    print(f"  {'─' * 40}")
    for name, arr in indicators.items():
        val = arr[last_idx]
        if np.isnan(val):
            print(f"  {name:<20} {'NaN':>15}")
        else:
            print(f"  {name:<20} {val:>15.4f}")


def run_atr_stop_backtest(df):
    print("\n" + "=" * 60)
    print("7. ATR 动态止损 + 风险回报比止盈")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    timestamps = df["time"].values.astype("int64")

    sma_fast = raptorbt.sma(close, period=10)
    sma_slow = raptorbt.sma(close, period=20)

    entries = ((sma_fast > sma_slow) & np.roll(sma_fast <= sma_slow, 1)).astype(bool)
    exits = ((sma_fast < sma_slow) & np.roll(sma_fast >= sma_slow, 1)).astype(bool)
    entries[:20] = False
    exits[:20] = False

    config = raptorbt.PyBacktestConfig(
        initial_capital=100000.0,
        fees=0.001,
        slippage=0.0005,
    )
    config.set_atr_stop(multiplier=2.0, period=14)
    config.set_risk_reward_target(ratio=2.0)

    result = raptorbt.run_single_backtest(
        timestamps=timestamps,
        open=open_, high=high, low=low, close=close,
        volume=volume,
        entries=entries, exits=exits,
        direction=1, weight=1.0, symbol=SYMBOL,
        config=config,
    )

    m = result.metrics
    print(f"  策略: SMA 交叉 + 2×ATR 止损 + 2:1 风险回报止盈")
    print(f"  总收益率: {m.total_return_pct:.2f}%  夏普: {m.sharpe_ratio:.2f}  "
          f"最大回撤: {m.max_drawdown_pct:.2f}%  交易数: {m.total_trades}  胜率: {m.win_rate_pct:.1f}%")

    trades = result.trades()
    exit_reasons = {}
    for t in trades:
        r = t.exit_reason
        exit_reasons[r] = exit_reasons.get(r, 0) + 1
    print(f"  出场原因分布: {exit_reasons}")
    return result


def run_basket_backtest(df):
    print("\n" + "=" * 60)
    print("8. 篮子回测 (多标的同步信号)")
    print("=" * 60)

    other_symbols = ["XAUUSD", "EURUSD", "GBPUSD"]
    dfs = {}
    for sym in other_symbols:
        try:
            date_to = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            date_from = (datetime.now(timezone.utc) - timedelta(days=BARS // 24 + 30)).strftime("%Y-%m-%d")
            data = api_get("/rates/from-date", params={
                "symbol": sym,
                "timeframe": TIMEFRAME,
                "date_from": date_from,
                "date_to": date_to,
            })
            rows = data.get("data", [])
            if not rows:
                print(f"  ⚠️ {sym} 无数据，跳过")
                continue
            sdf = pd.DataFrame(rows)
            sdf["time"] = pd.to_datetime(sdf["time"])
            sdf = sdf.sort_values("time").reset_index(drop=True)
            dfs[sym] = sdf
            print(f"  {sym}: 拉到 {len(sdf)} 根 K 线")
        except Exception as e:
            print(f"  ⚠️ {sym} 拉取失败: {e}")

    if len(dfs) < 2:
        print("  ⚠️ 可用品种不足 2 个，跳过篮子回测")
        return None

    min_len = min(len(d) for d in dfs.values())
    instruments = []
    instrument_configs = {}
    per_capital = 100000.0 / len(dfs)

    for i, (sym, sdf) in enumerate(dfs.items()):
        c = sdf["close"].values.astype(np.float64)[:min_len]
        o = sdf["open"].values.astype(np.float64)[:min_len]
        h = sdf["high"].values.astype(np.float64)[:min_len]
        l = sdf["low"].values.astype(np.float64)[:min_len]
        v = sdf["tick_volume"].values.astype(np.float64)[:min_len]
        ts = sdf["time"].values.astype("int64")[:min_len]

        sma_f = raptorbt.sma(c, period=10)
        sma_s = raptorbt.sma(c, period=20)
        ent = ((sma_f > sma_s) & np.roll(sma_f <= sma_s, 1)).astype(bool)
        ext = ((sma_f < sma_s) & np.roll(sma_f >= sma_s, 1)).astype(bool)
        ent[:20] = False
        ext[:20] = False

        instruments.append((ts, o, h, l, c, v, ent, ext, 1, 1.0 / len(dfs), sym))
        inst_cfg = raptorbt.PyInstrumentConfig(lot_size=0.01, alloted_capital=per_capital)
        instrument_configs[sym] = inst_cfg

    config = raptorbt.PyBacktestConfig(
        initial_capital=100000.0,
        fees=0.001,
        slippage=0.0005,
    )
    config.set_trailing_stop(0.03)

    for mode in ["any", "all"]:
        result = raptorbt.run_basket_backtest(
            instruments=instruments,
            config=config,
            sync_mode=mode,
            instrument_configs=instrument_configs,
        )
        m = result.metrics
        print(f"  sync_mode={mode:<8} → 收益: {m.total_return_pct:>7.2f}%  "
              f"夏普: {m.sharpe_ratio:>6.2f}  回撤: {m.max_drawdown_pct:>6.2f}%  "
              f"交易: {m.total_trades:>3d}  胜率: {m.win_rate_pct:>5.1f}%")
    return result


def run_pairs_backtest(df):
    print("\n" + "=" * 60)
    print("9. 配对交易 (EURUSD vs GBPUSD)")
    print("=" * 60)

    pairs = [("EURUSD", "GBPUSD"), ("EURUSD", "USDJPY")]
    for leg1_sym, leg2_sym in pairs:
        try:
            date_to = datetime.now(timezone.utc).strftime("%Y-%m-%d")
            date_from = (datetime.now(timezone.utc) - timedelta(days=BARS // 24 + 30)).strftime("%Y-%m-%d")

            data1 = api_get("/rates/from-date", params={
                "symbol": leg1_sym, "timeframe": TIMEFRAME,
                "date_from": date_from, "date_to": date_to,
            })
            data2 = api_get("/rates/from-date", params={
                "symbol": leg2_sym, "timeframe": TIMEFRAME,
                "date_from": date_from, "date_to": date_to,
            })

            rows1 = data1.get("data", [])
            rows2 = data2.get("data", [])
            if not rows1 or not rows2:
                print(f"  ⚠️ {leg1_sym}/{leg2_sym} 数据不足，跳过")
                continue

            df1 = pd.DataFrame(rows1)
            df2 = pd.DataFrame(rows2)
            df1["time"] = pd.to_datetime(df1["time"])
            df2["time"] = pd.to_datetime(df2["time"])
            df1 = df1.sort_values("time").reset_index(drop=True)
            df2 = df2.sort_values("time").reset_index(drop=True)

            min_len = min(len(df1), len(df2))
            c1 = df1["close"].values.astype(np.float64)[:min_len]
            c2 = df2["close"].values.astype(np.float64)[:min_len]

            spread = c1 - c2
            spread_sma = pd.Series(spread).rolling(20).mean().values
            spread_std = pd.Series(spread).rolling(20).std().values
            zscore = (spread - spread_sma) / np.where(spread_std > 0, spread_std, 1e-10)

            entries = (zscore < -2.0).astype(bool)
            exits = (np.abs(zscore) < 0.5).astype(bool)
            entries[:20] = False
            exits[:20] = False

            config = raptorbt.PyBacktestConfig(
                initial_capital=100000.0,
                fees=0.001,
                slippage=0.0005,
            )
            config.set_fixed_stop(0.03)

            result = raptorbt.run_pairs_backtest(
                leg1_timestamps=df1["time"].values.astype("int64")[:min_len],
                leg1_open=df1["open"].values.astype(np.float64)[:min_len],
                leg1_high=df1["high"].values.astype(np.float64)[:min_len],
                leg1_low=df1["low"].values.astype(np.float64)[:min_len],
                leg1_close=c1,
                leg1_volume=df1["tick_volume"].values.astype(np.float64)[:min_len],
                leg2_timestamps=df2["time"].values.astype("int64")[:min_len],
                leg2_open=df2["open"].values.astype(np.float64)[:min_len],
                leg2_high=df2["high"].values.astype(np.float64)[:min_len],
                leg2_low=df2["low"].values.astype(np.float64)[:min_len],
                leg2_close=c2,
                leg2_volume=df2["tick_volume"].values.astype(np.float64)[:min_len],
                entries=entries,
                exits=exits,
                direction=1,
                symbol=f"{leg1_sym}_{leg2_sym}",
                config=config,
                hedge_ratio=1.0,
            )

            m = result.metrics
            print(f"  {leg1_sym}/{leg2_sym} 配对 (Z-score 均值回归):")
            print(f"    收益: {m.total_return_pct:.2f}%  夏普: {m.sharpe_ratio:.2f}  "
                  f"回撤: {m.max_drawdown_pct:.2f}%  交易: {m.total_trades}  胜率: {m.win_rate_pct:.1f}%")
        except Exception as e:
            print(f"  ⚠️ {leg1_sym}/{leg2_sym} 配对失败: {e}")
    return None


def run_monte_carlo(result1, result2):
    print("\n" + "=" * 60)
    print("10. 蒙特卡洛组合模拟")
    print("=" * 60)

    returns1 = result1.returns()
    returns2 = result2.returns()
    min_len = min(len(returns1), len(returns2))

    r1 = returns1[:min_len]
    r2 = returns2[:min_len]

    mask = ~(np.isnan(r1) | np.isnan(r2))
    r1 = r1[mask]
    r2 = r2[mask]

    if len(r1) < 50:
        print("  ⚠️ 有效收益率数据不足，跳过蒙特卡洛模拟")
        return

    corr = np.corrcoef(r1, r2)[0, 1]
    correlation_matrix = [
        np.array([1.0, corr]),
        np.array([corr, 1.0]),
    ]

    print(f"  策略 1 (SMA交叉) vs 策略 2 (多策略) 收益率相关系数: {corr:.3f}")

    for n_sim in [1000, 10000]:
        mc_result = raptorbt.simulate_portfolio_mc(
            returns=[r1, r2],
            weights=np.array([0.6, 0.4]),
            correlation_matrix=correlation_matrix,
            initial_value=100000.0,
            n_simulations=n_sim,
            horizon_days=252,
            seed=42,
        )

        print(f"\n  模拟次数: {n_sim:,}")
        print(f"  {'─' * 40}")
        print(f"  预期收益:       {mc_result['expected_return']:>10.2f} %")
        print(f"  亏损概率:       {mc_result['probability_of_loss']:>10.2%}")
        print(f"  VaR (95%):      {mc_result['var_95']:>10.2f} %")
        print(f"  CVaR (95%):     {mc_result['cvar_95']:>10.2f} %")

        print(f"  分位数路径终值:")
        for pct, path in mc_result["percentile_paths"]:
            print(f"    P{pct:>2.0f}: {path[-1]:>12,.2f}")


def export_results(result, df, strategy_name="sma_cross"):
    print("\n" + "=" * 60)
    print("11. 导出回测结果")
    print("=" * 60)

    output_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backtest_output")
    os.makedirs(output_dir, exist_ok=True)

    trades = result.trades()
    if trades:
        rows = []
        for t in trades:
            rows.append({
                "trade_id": t.id,
                "symbol": t.symbol,
                "direction": "Long" if t.direction == 1 else "Short",
                "entry_idx": t.entry_idx,
                "exit_idx": t.exit_idx,
                "entry_time": df["time"].iloc[t.entry_idx] if t.entry_idx < len(df) else "",
                "exit_time": df["time"].iloc[t.exit_idx] if t.exit_idx < len(df) else "",
                "entry_price": t.entry_price,
                "exit_price": t.exit_price,
                "size": t.size,
                "pnl": t.pnl,
                "return_pct": t.return_pct,
                "fees": t.fees,
                "exit_reason": t.exit_reason,
            })
        trades_df = pd.DataFrame(rows)
        trades_path = os.path.join(output_dir, f"{strategy_name}_trades.csv")
        trades_df.to_csv(trades_path, index=False, encoding="utf-8-sig")
        print(f"  交易记录 → {trades_path}  ({len(trades_df)} 笔)")

    equity = result.equity_curve()
    drawdown = result.drawdown_curve()
    returns = result.returns()

    curve_df = pd.DataFrame({
        "time": df["time"].values[:len(equity)],
        "equity": equity,
        "drawdown": drawdown,
        "returns": returns,
    })
    curve_path = os.path.join(output_dir, f"{strategy_name}_curves.csv")
    curve_df.to_csv(curve_path, index=False, encoding="utf-8-sig")
    print(f"  权益曲线 → {curve_path}  ({len(curve_df)} 根)")

    m = result.metrics
    metrics_dict = m.to_dict()
    metrics_dict["total_trades"] = m.total_trades
    metrics_dict["total_closed_trades"] = m.total_closed_trades
    metrics_dict["total_open_trades"] = m.total_open_trades
    metrics_dict["winning_trades"] = m.winning_trades
    metrics_dict["losing_trades"] = m.losing_trades
    metrics_dict["max_consecutive_wins"] = m.max_consecutive_wins
    metrics_dict["max_consecutive_losses"] = m.max_consecutive_losses
    metrics_dict["avg_holding_period"] = m.avg_holding_period
    metrics_dict["avg_winning_duration"] = m.avg_winning_duration
    metrics_dict["avg_losing_duration"] = m.avg_losing_duration
    metrics_dict["start_value"] = m.start_value
    metrics_dict["end_value"] = m.end_value
    metrics_dict["total_fees_paid"] = m.total_fees_paid
    metrics_dict["open_trade_pnl"] = m.open_trade_pnl
    metrics_dict["exposure_pct"] = m.exposure_pct
    metrics_dict["payoff_ratio"] = m.payoff_ratio
    metrics_dict["recovery_factor"] = m.recovery_factor
    metrics_dict["omega_ratio"] = m.omega_ratio

    metrics_df = pd.DataFrame(list(metrics_dict.items()), columns=["metric", "value"])
    metrics_path = os.path.join(output_dir, f"{strategy_name}_metrics.csv")
    metrics_df.to_csv(metrics_path, index=False, encoding="utf-8-sig")
    print(f"  绩效指标 → {metrics_path}  ({len(metrics_df)} 项)")

    print(f"\n  所有文件保存在: {output_dir}")


def test_ferro_indicators(df):
    print("\n" + "=" * 60)
    print("7. ferro-ta 新指标全量测试 (45+ 指标)")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    n = len(close)
    errors = []
    tested = 0

    def check(name, arr, expect_len=None, expect_finite=True, value_range=None, min_valid=1):
        nonlocal tested
        tested += 1
        if expect_len is not None and len(arr) != expect_len:
            errors.append(f"{name}: 长度 {len(arr)} != 期望 {expect_len}")
            return
        valid = arr[~np.isnan(arr)]
        if len(valid) < min_valid:
            errors.append(f"{name}: 有效值仅 {len(valid)} 个 (最少需要 {min_valid})")
            return
        if expect_finite:
            inf_count = np.isinf(valid).sum()
            if inf_count > 0:
                errors.append(f"{name}: 有 {inf_count} 个 inf 值")
        if value_range and len(valid) > 0:
            lo, hi = value_range
            out_of_range = ((valid < lo) | (valid > hi)).sum()
            if out_of_range > 0:
                errors.append(f"{name}: {out_of_range} 个值超出 [{lo}, {hi}]")
        last = arr[-1]
        status = f"{last:.4f}" if not np.isnan(last) else "NaN"
        print(f"  {name:<25} 长度={len(arr):>4}  有效值={len(valid):>4}  最新={status}")

    # ── Overlap 均线类 ──
    print("\n  ── Overlap 均线类 ──")
    check("WMA(20)", raptorbt.wma(close, period=20), n, min_valid=10)
    check("DEMA(20)", raptorbt.dema(close, period=20), n, min_valid=10)
    check("TEMA(20)", raptorbt.tema(close, period=20), n, min_valid=10)
    check("KAMA(10)", raptorbt.kama(close, period=10), n, min_valid=10)
    check("T3(20)", raptorbt.t3(close, period=20, vfactor=0.7), n, min_valid=10)
    check("TRIMA(20)", raptorbt.trima(close, period=20), n, min_valid=10)
    check("Midpoint(20)", raptorbt.midpoint(close, period=20), n, min_valid=10)
    check("Midprice(20)", raptorbt.midprice(high, low, period=20), n, min_valid=10)
    check("SAR(0.02,0.2)", raptorbt.sar(high, low, acceleration=0.02, maximum=0.2), n, min_valid=10)

    # ── Momentum 动量类 ──
    print("\n  ── Momentum 动量类 ──")
    check("CCI(20)", raptorbt.cci(high, low, close, period=20), n, min_valid=10)
    check("WillR(14)", raptorbt.willr(high, low, close, period=14), n, value_range=(-101, 1), min_valid=10)
    check("ROC(10)", raptorbt.roc(close, period=10), n, min_valid=10)
    check("MOM(10)", raptorbt.mom(close, period=10), n, min_valid=10)
    check("CMO(14)", raptorbt.cmo(close, period=14), n, min_valid=10)
    check("TRIX(12)", raptorbt.trix(close, period=12), n, min_valid=10)
    check("UltOSC(7,14,28)", raptorbt.ultosc(high, low, close, period1=7, period2=14, period3=28), n, value_range=(-1, 101), min_valid=10)
    check("AroonOsc(14)", raptorbt.aroonosc(high, low, period=14), n, value_range=(-101, 101), min_valid=10)
    check("BOP", raptorbt.bop(open_, high, low, close), n, value_range=(-1.01, 1.01), min_valid=10)
    check("Plus_DI(14)", raptorbt.plus_di(high, low, close, period=14), n, value_range=(-1, 101), min_valid=10)
    check("Minus_DI(14)", raptorbt.minus_di(high, low, close, period=14), n, value_range=(-1, 101), min_valid=10)

    # ── 多输出动量指标 ──
    print("\n  ── 多输出动量指标 ──")
    try:
        stochrsi_k, stochrsi_d = raptorbt.stochrsi(close, timeperiod=14, fastk_period=5, fastd_period=3)
        check("StochRSI_K", stochrsi_k, n, value_range=(-5, 105), min_valid=10)
        check("StochRSI_D", stochrsi_d, n, value_range=(-5, 105), min_valid=10)
    except Exception as e:
        errors.append(f"StochRSI: {e}")
        tested += 2

    try:
        adx_val, plus_di, minus_di = raptorbt.adx_all(high, low, close, period=14)
        check("ADX_all(14)", adx_val, n, value_range=(-1, 101), min_valid=10)
        check("ADX_all +DI", plus_di, n, value_range=(-1, 101), min_valid=10)
        check("ADX_all -DI", minus_di, n, value_range=(-1, 101), min_valid=10)
    except Exception as e:
        errors.append(f"ADX_all: {e}")
        tested += 3

    try:
        aroon_up, aroon_down = raptorbt.aroon(high, low, period=14)
        check("Aroon_Up", aroon_up, n, value_range=(-1, 101), min_valid=10)
        check("Aroon_Down", aroon_down, n, value_range=(-1, 101), min_valid=10)
    except Exception as e:
        errors.append(f"Aroon: {e}")
        tested += 2

    try:
        ppo_line, ppo_signal, ppo_hist = raptorbt.ppo(close, fastperiod=12, slowperiod=26, signalperiod=9)
        check("PPO_Line", ppo_line, n, min_valid=10)
        check("PPO_Signal", ppo_signal, n, min_valid=10)
        check("PPO_Hist", ppo_hist, n, min_valid=10)
    except Exception as e:
        errors.append(f"PPO: {e}")
        tested += 3

    # ── Volatility 波动率类 ──
    print("\n  ── Volatility 波动率类 ──")
    check("NATR(14)", raptorbt.natr(high, low, close, period=14), n, min_valid=10)
    check("TRange", raptorbt.trange(high, low, close), n, min_valid=10)
    check("StdDev(20)", raptorbt.stddev(close, period=20, nbdev=1.0), n, min_valid=10)
    check("VAR(20)", raptorbt.var(close, period=20, nbdev=1.0), n, min_valid=10)

    # ── Volume 成交量类 ──
    print("\n  ── Volume 成交量类 ──")
    check("AD", raptorbt.ad(high, low, close, volume), n, min_valid=10)
    check("ADOSC(3,10)", raptorbt.adosc(high, low, close, volume, fastperiod=3, slowperiod=10), n, min_valid=10)
    check("OBV", raptorbt.obv(close, volume), n, min_valid=10)
    check("MFI(14)", raptorbt.mfi(high, low, close, volume, period=14), n, value_range=(-1, 101), min_valid=10)

    # ── Price Transform 价格变换类 ──
    print("\n  ── Price Transform 价格变换类 ──")
    check("TypPrice", raptorbt.typprice(high, low, close), n, min_valid=10)
    check("MedPrice", raptorbt.medprice(high, low), n, min_valid=10)
    check("AvgPrice", raptorbt.avgprice(open_, high, low, close), n, min_valid=10)
    check("WclPrice", raptorbt.wclprice(high, low, close), n, min_valid=10)

    # ── Statistic 统计类 ──
    print("\n  ── Statistic 统计类 ──")
    check("LinearReg(20)", raptorbt.linearreg(close, period=20), n, min_valid=10)
    check("LinReg_Slope(20)", raptorbt.linearreg_slope(close, period=20), n, min_valid=10)
    check("LinReg_Angle(20)", raptorbt.linearreg_angle(close, period=20), n, min_valid=10)
    check("LinReg_Intercept(20)", raptorbt.linearreg_intercept(close, period=20), n, min_valid=10)
    check("TSF(20)", raptorbt.tsf(close, period=20), n, min_valid=10)
    check("Beta(20)", raptorbt.beta(close, close, period=20), n, min_valid=10)
    check("Correl(20)", raptorbt.correl(close, close, period=20), n, min_valid=10)

    # ── ADXr ──
    print("\n  ── ADXr ──")
    check("ADXr(14)", raptorbt.adxr(high, low, close, period=14), n, value_range=(-1, 101), min_valid=10)

    # ── APO ──
    print("\n  ── APO ──")
    check("APO(12,26)", raptorbt.apo(close, fastperiod=12, slowperiod=26), n, min_valid=10)

    # ── 汇总 ──
    print("\n" + "─" * 60)
    if errors:
        print(f"  ❌ 测试 {tested} 个指标，发现 {len(errors)} 个错误:")
        for e in errors:
            print(f"    - {e}")
    else:
        print(f"  ✅ 全部 {tested} 个指标测试通过，无错误！")
    print("─" * 60)
    return len(errors) == 0


def run_ferro_strategy_backtest(df):
    print("\n" + "=" * 60)
    print("8. ferro-ta 指标策略回测 (SAR + ADX + CCI 组合)")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    timestamps = df["time"].values.astype("int64")

    sar_val = raptorbt.sar(high, low, acceleration=0.02, maximum=0.2)
    adx_val, plus_di, minus_di = raptorbt.adx_all(high, low, close, period=14)
    cci_val = raptorbt.cci(high, low, close, period=20)

    sar_above = sar_val > close
    sar_below = sar_val < close

    adx_strong = adx_val > 25
    bullish_di = plus_di > minus_di
    bearish_di = minus_di > plus_di

    cci_oversold = cci_val < -100
    cci_overbought = cci_val > 100

    entries_long = sar_below & adx_strong & bullish_di & cci_oversold
    entries_short = sar_above & adx_strong & bearish_di & cci_overbought
    entries = entries_long | entries_short

    exits = (sar_below & bearish_di & adx_strong) | (sar_above & bullish_di & adx_strong)

    warmup = 40
    entries[:warmup] = False
    exits[:warmup] = False

    n_long = int(entries_long.sum())
    n_short = int(entries_short.sum())
    n_exits = int(exits.sum())
    print(f"  SAR+ADX+CCI 组合信号: {n_long} 次做多, {n_short} 次做空, {n_exits} 次出场")

    config = raptorbt.PyBacktestConfig(
        initial_capital=100000.0,
        fees=0.001,
        slippage=0.0005,
    )
    config.set_atr_stop(multiplier=2.5, period=14)
    config.set_fixed_target(0.04)

    result = raptorbt.run_single_backtest(
        timestamps=timestamps,
        open=open_, high=high, low=low, close=close,
        volume=volume,
        entries=entries, exits=exits,
        direction=1, weight=1.0, symbol=SYMBOL,
        config=config,
    )

    m = result.metrics
    print(f"\n  {'─' * 40}")
    print(f"  SAR+ADX+CCI 策略回测结果")
    print(f"  {'─' * 40}")
    print(f"  总收益率:     {m.total_return_pct:>10.2f} %")
    print(f"  夏普比率:     {m.sharpe_ratio:>10.2f}")
    print(f"  最大回撤:     {m.max_drawdown_pct:>10.2f} %")
    print(f"  总交易数:     {m.total_trades:>10d}")
    print(f"  胜率:         {m.win_rate_pct:>10.1f} %")
    print(f"  盈利因子:     {m.profit_factor:>10.2f}")

    trades = result.trades()
    if trades:
        exit_reasons = {}
        for t in trades:
            r = t.exit_reason
            exit_reasons[r] = exit_reasons.get(r, 0) + 1
        print(f"  出场原因分布: {exit_reasons}")

    return result


def test_extended_indicators(df):
    """P0 批 + Hilbert + 市场状态 + 投资组合工具 全量测试"""
    print("\n" + "=" * 60)
    print("12. P0 扩展指标全量测试")
    print("=" * 60)

    close = df["close"].values.astype(np.float64)
    open_ = df["open"].values.astype(np.float64)
    high = df["high"].values.astype(np.float64)
    low = df["low"].values.astype(np.float64)
    volume = df["tick_volume"].values.astype(np.float64)
    n = len(close)
    errors = []

    def check(name, arr, expect_len=None, value_range=None, min_valid=1, dtype=None):
        if expect_len is not None and len(arr) != expect_len:
            errors.append(f"{name}: 长度 {len(arr)} != 期望 {expect_len}")
            return
        valid = arr[~np.isnan(arr)]
        if len(valid) < min_valid:
            errors.append(f"{name}: 有效值仅 {len(valid)} 个 (最少需要 {min_valid})")
            return
        if value_range and len(valid) > 0:
            lo, hi = value_range
            out = ((valid < lo) | (valid > hi)).sum()
            if out > 0:
                errors.append(f"{name}: {out} 个值超出 [{lo}, {hi}]")
        if dtype is not None and not isinstance(arr, dtype):
            errors.append(f"{name}: dtype 应为 {dtype.__name__} 而非 {type(arr).__name__}")
        last = arr[-1]
        status = f"{last:.4f}" if not np.isnan(last) else "NaN"
        print(f"  {name:<30} 长度={len(arr):>4}  有效值={len(valid):>4}  最新={status}")

    def check_tuple(name, arrs, expect_len=None, value_ranges=None, min_valids=None):
        """测试多返回值的指标"""
        if not min_valids:
            min_valids = [1] * len(arrs)
        if not value_ranges:
            value_ranges = [None] * len(arrs)
        for i, (a, vmin, vr) in enumerate(zip(arrs, min_valids, value_ranges)):
            sub = check(f"{name}[{i}]", a, expect_len, vr, vmin)

    # ── P0 扩展指标 ──
    print("\n  ── P0 扩展指标 ──")
    check("VWMA(20)", raptorbt.vwma(close, volume, period=20), n, min_valid=10)
    donch = raptorbt.donchian(high, low, period=20)
    check("Donchian(20) upper", donch[0], n, min_valid=10)
    check("Donchian(20) middle", donch[1], n, min_valid=10)
    check("Donchian(20) lower", donch[2], n, min_valid=10)
    # Donchian 的上下边界应当包含收盘价
    donch_upper = donch[0]
    donch_lower = donch[2]
    valid_idx = ~np.isnan(donch_upper) & ~np.isnan(donch_lower)
    if valid_idx.sum() > 0:
        bad = ((donch_upper[valid_idx] < close[valid_idx]) | (donch_lower[valid_idx] > close[valid_idx])).sum()
        if bad > 0:
            errors.append(f"Donchian: {bad} 根 K 线收盘价落在通道之外")

    ci_val = raptorbt.choppiness_index(high, low, close, period=14)
    check("Choppiness(14)", ci_val, n, value_range=(-1, 101), min_valid=10)

    check("HullMA(14)", raptorbt.hull_ma(close, period=14), n, min_valid=10)

    cl, cs = raptorbt.chandelier_exit(high, low, close, period=22, multiplier=3.0)
    check("Chandelier long_exit", cl, n, min_valid=10)
    check("Chandelier short_exit", cs, n, min_valid=10)
    # Note: long_exit can be < short_exit in sharp trending markets — this is valid.
    # The real check is that both are finite and reasonably close to price.

    tenkan, kijun, senkou_a, senkou_b, chikou = raptorbt.ichimoku(
        high, low, close, tenkan_period=9, kijun_period=26, senkou_b_period=52, displacement=26)
    check("Ichimoku tenkan", tenkan, n, min_valid=10)
    check("Ichimoku kijun", kijun, n, min_valid=10)
    check("Ichimoku senkou_a", senkou_a, n, min_valid=10)
    check("Ichimoku senkou_b", senkou_b, n, min_valid=10)
    check("Ichimoku chikou", chikou, n, min_valid=10)

    pivot, r1, s1, r2, s2 = raptorbt.pivot_points(high, low, close, method="classic")
    check("Pivot pivot", pivot, n, min_valid=5)
    check("Pivot R1", r1, n, min_valid=5)
    check("Pivot S1", s1, n, min_valid=5)
    check("Pivot R2", r2, n, min_valid=5)
    check("Pivot S2", s2, n, min_valid=5)
    # R2 > R1 > pivot > S1 > S2 对于有效值
    valid_pp = ~np.isnan(pivot) & ~np.isnan(r1) & ~np.isnan(s1) & ~np.isnan(r2) & ~np.isnan(s2)
    if valid_pp.sum() > 0:
        order = (r2[valid_pp] > r1[valid_pp]) & (r1[valid_pp] > pivot[valid_pp]) & \
                (pivot[valid_pp] > s1[valid_pp]) & (s1[valid_pp] > s2[valid_pp])
        bad = (~order).sum()
        if bad > 0:
            errors.append(f"Pivot: {bad} 处 R2>R1>PIVOT>S1>S2 排序异常")

    # ── Hilbert Transform ──
    print("\n  ── Hilbert Transform (周期变换) ──")
    check("HT_Trendline", raptorbt.ht_trendline(close), n, min_valid=10)
    check("HT_DCPeriod", raptorbt.ht_dcperiod(close), n, value_range=(-1, 100), min_valid=10)
    check("HT_DCPhase", raptorbt.ht_dcphase(close), n, value_range=(-361, 361), min_valid=10)
    ip, quad = raptorbt.ht_phasor(close)
    check("HT_Phasor in_phase", ip, n, min_valid=10)
    check("HT_Phasor quadrature", quad, n, min_valid=10)
    sin, lead = raptorbt.ht_sine(close)
    check("HT_Sine", sin, n, value_range=(-1.05, 1.05), min_valid=10)
    check("HT_LeadSine", lead, n, value_range=(-1.05, 1.05), min_valid=10)
    mode = raptorbt.ht_trendmode(close)
    # mode 是 i32 数组
    check("HT_TrendMode", mode, n, value_range=(-2, 2), min_valid=10)
    # 验证 TrendMode 只有 0/1
    mode_valid = mode[~np.isnan(mode)] if mode.dtype.kind == 'f' else mode
    unique = np.unique(mode_valid)
    bad = np.isin(unique, [0, 1]).all() == False and len(unique) > 0
    if len(unique) > 0 and not set(unique) <= {0, 1}:
        errors.append(f"HT_TrendMode: 含有非 0/1 值: {unique}")

    # ── 市场状态检测 ──
    print("\n  ── 市场状态检测 ──")
    adx_raw = raptorbt.adx(high, low, close, period=14)
    r_adx = raptorbt.regime_adx(adx_raw, threshold=25.0)
    check("Regime_ADX", r_adx, n, value_range=(-2, 2), min_valid=5)
    # 验证只有 -1/0/1
    unique_r = np.unique(r_adx)
    if not set(unique_r) <= {-1, 0, 1}:
        errors.append(f"Regime_ADX: 含有非 -1/0/1 值: {unique_r}")

    atr_raw = raptorbt.atr(high, low, close, period=14)
    r_combo = raptorbt.regime_combined(adx_raw, atr_raw, close, 25.0, 2.0)
    check("Regime_Combined", r_combo, n, value_range=(-2, 2), min_valid=5)
    unique_c = np.unique(r_combo)
    if not set(unique_c) <= {-1, 0, 1}:
        errors.append(f"Regime_Combined: 含有非 -1/0/1 值: {unique_c}")

    breaks = raptorbt.detect_breaks_cusum(close, window=30, threshold=1.5, slack=0.5)
    check("Breaks_CUSUM", breaks, n, value_range=(-2, 2), min_valid=5)
    unique_b = np.unique(breaks)
    if not set(unique_b) <= {-1, 0, 1}:
        errors.append(f"Breaks_CUSUM: 含有非 -1/0/1 值: {unique_b}")

    vol_breaks = raptorbt.rolling_variance_break(close, short_window=10, long_window=30, threshold=2.0)
    check("Breaks_VarRatio", vol_breaks, n, value_range=(-2, 2), min_valid=5)
    unique_v = np.unique(vol_breaks)
    if not set(unique_v) <= {-1, 0, 1}:
        errors.append(f"Breaks_VarRatio: 含有非 -1/0/1 值: {unique_v}")

    # ── 投资组合工具 ──
    print("\n  ── 投资组合工具 ──")
    # 创建第二组数据（随机偏移模拟）
    close2 = close * (1 + np.random.uniform(-0.01, 0.01, n))
    rb = raptorbt.rolling_beta(close, close2, window=20)
    check("Rolling_Beta", rb, n, min_valid=10)

    dd_series, max_dd = raptorbt.drawdown_series(close)
    check("DrawdownSeries", dd_series, n, value_range=(-1.01, 0.01), min_valid=5)
    if max_dd > 0:
        errors.append(f"DrawdownSeries max_dd 应为非正值: {max_dd}")

    check("ZScore(20)", raptorbt.zscore_series(close, window=20), n, min_valid=10)

    returns = np.diff(close) / close[:-1]
    returns2 = np.diff(close2) / close2[:-1]
    rs = raptorbt.relative_strength(returns, returns2)
    check("RelativeStrength", rs, None, min_valid=10)

    hedge = 1.0
    spr = raptorbt.spread(close, close2, hedge)
    check("Spread", spr, n, min_valid=10)
    # 验证 spread ≈ close - close2
    valid_sp = ~np.isnan(spr) & ~np.isnan(close - close2)
    if valid_sp.sum() > 0:
        diff = np.abs(spr[valid_sp] - (close[valid_sp] - close2[valid_sp]))
        bad = (diff > 1e-6).sum()
        if bad > 0:
            errors.append(f"Spread: {bad} 处与 close - close2 偏差超过 1e-6")

    r = raptorbt.ratio(close, close2)
    check("Ratio", r, n, min_valid=10)
    # 验证 ratio ≈ close / close2
    valid_r = ~np.isnan(r) & ~np.isnan(close / close2)
    if valid_r.sum() > 0:
        diff_r = np.abs(r[valid_r] - close[valid_r] / close2[valid_r])
        bad_r = (diff_r > 1e-6).sum()
        if bad_r > 0:
            errors.append(f"Ratio: {bad_r} 处与 close / close2 偏差超过 1e-6")

    # ── 汇总 ──
    print("\n" + "─" * 60)
    if errors:
        print(f"  ❌ 测试发现 {len(errors)} 个错误:")
        for e in errors:
            print(f"    - {e}")
    else:
        print(f"  ✅ 全部扩展指标测试通过，无错误！")
    print("─" * 60)
    return len(errors) == 0


def main():
    print("╔══════════════════════════════════════════════════════════╗")
    print("║   RaptorBT + Mt5Bridge 集成测试                         ║")
    print("║   从 MT5 拉取真实数据 → RaptorBT 回测                   ║")
    print("╚══════════════════════════════════════════════════════════╝\n")

    check_health()
    fetch_account()

    df = fetch_klines()

    result1 = run_backtest(df)
    result2 = run_multi_indicator_backtest(df)
    show_indicators(df)
    test_ferro_indicators(df)
    test_extended_indicators(df)
    result_ferro = run_ferro_strategy_backtest(df)
    result3 = run_atr_stop_backtest(df)
    run_basket_backtest(df)
    run_pairs_backtest(df)
    run_monte_carlo(result1, result2)

    export_results(result1, df, strategy_name="sma_cross")
    export_results(result2, df, strategy_name="multi_strategy")
    if result3:
        export_results(result3, df, strategy_name="atr_stop_rr")
    if result_ferro:
        export_results(result_ferro, df, strategy_name="ferro_sar_adx_cci")

    print("\n" + "=" * 60)
    print("✅ 全部测试完成！")
    print("=" * 60)


if __name__ == "__main__":
    main()