Dex Quant Skill

*.pyc
__pycache__/
.env
*.egg-info/
dist/
build/
.history/
__pycache__/

display_name: "DEX Quant Skill"
short_description: "AI 量化交易平台 — 策略设计→回测→评审→信号→执行"
default_prompt: "帮我设计一个加密货币量化策略"

{
  "review_id": "",
  "strategy_id": "",
  "backtest_id": "",
  "version": "",
  "reviewer": "backtest-reviewer",
  "review_time": "",
  "decision": "pending",
  "summary": "",

  "strengths": [],
  "risks": [],

  "metrics_summary": {
    "sharpe_ratio": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "max_drawdown": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "win_rate": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "profit_factor": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "calmar_ratio": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "sortino_ratio": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "total_return": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "annual_return": {
      "value": null,
      "rating": "",
      "comment": ""
    },
    "total_trades": {
      "value": null,
      "rating": "",
      "comment": ""
    }
  },

  "overfitting_assessment": {
    "risk_level": "",
    "freedom_ratio": null,
    "oos_decay": null,
    "parameter_sensitivity": "",
    "detail": ""
  },

  "cost_impact": {
    "pre_cost_return": null,
    "post_cost_return": null,
    "erosion_ratio": null,
    "double_cost_still_profitable": null,
    "detail": ""
  },

  "revenue_source_analysis": {
    "top5_concentration": null,
    "direction_balance": "",
    "regime_dependency": "",
    "detail": ""
  },

  "market_regime_analysis": {
    "trend_performance": "",
    "ranging_performance": "",
    "crash_performance": "",
    "detail": ""
  },

  "tradability_assessment": {
    "liquidity_ok": null,
    "latency_sensitive": null,
    "funding_rate_impact": "",
    "detail": ""
  },

  "score": null,
  "score_breakdown": {
    "sharpe_score": null,
    "drawdown_score": null,
    "expectancy_score": null,
    "calmar_score": null,
    "sortino_score": null,
    "cost_score": null,
    "liquidation_score": null
  },

  "required_actions": [],
  "recommended_next_step": "",

  "metadata": {
    "review_checklist_version": "v1.0",
    "metrics_guide_version": "v1.0",
    "notes": ""
  }
}

"""
DEX Quant Server API 客户端 — 信号驱动架构（含 Token 认证）

Skill 端调用流程:
  1. strategy-maker 生成策略脚本
  2. 本地运行脚本，拿到信号列表
  3. 调 run_backtest() 把信号发给 Server（自动携带 Token）
  4. Server 拉 K 线（带缓存）+ 回测引擎回放信号
  5. 返回绩效结果，展示给用户

认证:
  - 首次使用自动注册机器码，获取 Token（免费 3 个策略配额）
  - Token 缓存在 ~/.dex-quant/config.json
  - 所有请求自动携带 X-Token 头

用法:
    client = QuantAPIClient("http://your-server:8000")
    result = client.run_backtest(
        strategy_name="BTC MACD 策略",
        symbol="BTCUSDT",
        timeframe="1h",
        start_date="2024-01-01",
        end_date="2024-12-31",
        signals=[...],
    )
    client.print_metrics(result)
"""

from __future__ import annotations

from typing import Optional

import httpx
from loguru import logger

from machine_auth import MachineAuth

DEFAULT_SERVER_URL = "https://quant.qa1.dex.hashkeydev.com"
API_PREFIX = "/api/v1"


class QuantAPIClient:
    """DEX Quant Server HTTP 客户端（自动认证）"""

    def __init__(self, server_url: str = DEFAULT_SERVER_URL, timeout: float = 300.0):
        self.server_url = server_url
        self.base_url = server_url.rstrip("/") + API_PREFIX
        self._client = httpx.Client(timeout=timeout)

        self._auth = MachineAuth(server_url)
        self._token = self._auth.register_or_load()

    def _headers(self) -> dict:
        return {"X-Token": self._token}

    # ═══════════════ 回测 ═══════════════

    def run_backtest(
        self,
        strategy_name: str,
        symbol: str,
        timeframe: str,
        start_date: str,
        end_date: str,
        signals: list[dict],
        strategy_id: str = "",
        initial_capital: float = 100_000.0,
        leverage: int = 1,
        fee_rate: float = 0.0005,
        slippage_bps: float = 5.0,
        margin_mode: str = "isolated",
        direction: str = "long_short",
    ) -> dict:
        """
        提交信号驱动回测。

        参数:
            strategy_name: 策略名称
            symbol: 交易对 (BTCUSDT)
            timeframe: K 线周期 (15m / 1h / 2h / 1d)
            start_date: 开始日期 "YYYY-MM-DD"
            end_date: 结束日期 "YYYY-MM-DD"
            signals: 信号列表，每个信号包含:
                timestamp, symbol, action (buy/sell/close),
                direction (long/short), confidence, reason,
                price_at_signal, suggested_stop_loss, suggested_take_profit

        返回:
            BacktestResponse 字典:
                backtest_id, status, metrics, trades, equity_curve, conclusion
        """
        payload = {
            "strategy_name": strategy_name,
            "strategy_id": strategy_id,
            "symbol": symbol,
            "timeframe": timeframe,
            "start_date": start_date,
            "end_date": end_date,
            "signals": signals,
            "initial_capital": initial_capital,
            "leverage": leverage,
            "fee_rate": fee_rate,
            "slippage_bps": slippage_bps,
            "margin_mode": margin_mode,
            "direction": direction,
        }

        logger.info(
            "提交回测 | {} {} {} | {} → {} | {} 个信号",
            strategy_name, symbol, timeframe, start_date, end_date, len(signals),
        )
        resp = self._client.post(f"{self.base_url}/backtest/run", json=payload, headers=self._headers())
        resp.raise_for_status()
        result = resp.json()

        status = result.get("status", "unknown")
        if status == "completed":
            metrics = result.get("metrics", {})
            logger.info(
                "回测完成 | 收益={:.2%} | Sharpe={:.2f} | 回撤={:.2%} | "
                "交易={} | 结论={}",
                metrics.get("total_return_pct", 0),
                metrics.get("sharpe_ratio", 0),
                abs(metrics.get("max_drawdown_pct", 0)),
                metrics.get("total_trades", 0),
                result.get("conclusion", ""),
            )
        else:
            logger.error("回测失败 | {}", result.get("error"))

        return result

    def get_backtest(self, backtest_id: str) -> dict:
        """查询已保存的回测结果"""
        resp = self._client.get(f"{self.base_url}/backtest/{backtest_id}", headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    def get_trades(self, backtest_id: str) -> dict:
        """获取回测交易记录"""
        resp = self._client.get(f"{self.base_url}/backtest/{backtest_id}/trades", headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    def get_equity(self, backtest_id: str) -> dict:
        """获取权益曲线"""
        resp = self._client.get(f"{self.base_url}/backtest/{backtest_id}/equity", headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    # ═══════════════ 数据 ═══════════════

    def get_klines(
        self,
        symbol: str,
        interval: str,
        start_date: str,
        end_date: str,
        exchange: str = "binance",
    ) -> list[dict]:
        """获取 K 线数据（Server 端带缓存，同币同周期不重复下载）"""
        payload = {
            "symbol": symbol,
            "interval": interval,
            "start_date": start_date,
            "end_date": end_date,
            "exchange": exchange,
        }
        resp = self._client.post(f"{self.base_url}/data/klines", json=payload, headers=self._headers())
        resp.raise_for_status()
        result = resp.json()
        logger.info("K线 | {} {} | {} 条", symbol, interval, result.get("rows", 0))
        return result.get("data", [])

    def list_symbols(self, exchange: str = "binance") -> list[str]:
        """列出可用交易对"""
        resp = self._client.get(f"{self.base_url}/data/symbols", params={"exchange": exchange}, headers=self._headers())
        resp.raise_for_status()
        return resp.json().get("symbols", [])

    # ═══════════════ 策略 ═══════════════

    def save_strategy(
        self,
        name: str,
        script_content: str = "",
        description: str = "",
        symbol: str = "BTCUSDT",
        timeframe: str = "1h",
        direction: str = "long_short",
        version: str = "v1.0",
        tags: list[str] = None,
    ) -> dict:
        """保存策略到 Server（含脚本源码）"""
        payload = {
            "name": name,
            "description": description,
            "script_content": script_content,
            "symbol": symbol,
            "timeframe": timeframe,
            "direction": direction,
            "version": version,
            "tags": tags or [],
        }
        resp = self._client.post(f"{self.base_url}/strategies/", json=payload, headers=self._headers())
        resp.raise_for_status()
        result = resp.json()
        logger.info("策略已保存 | {} ({})", name, result.get("strategy_id", ""))
        return result

    def list_strategies(self) -> list[dict]:
        """列出所有策略"""
        resp = self._client.get(f"{self.base_url}/strategies/", headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    def get_strategy(self, strategy_id: str) -> dict:
        """获取策略详情（含脚本源码）"""
        resp = self._client.get(f"{self.base_url}/strategies/{strategy_id}", headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    # ═══════════════ 信号 ═══════════════

    def save_signals(self, strategy_id: str, signals: list[dict]) -> dict:
        """批量保存信号到 Server"""
        resp = self._client.post(
            f"{self.base_url}/signals/batch",
            json=signals,
            params={"strategy_id": strategy_id},
            headers=self._headers(),
        )
        resp.raise_for_status()
        return resp.json()

    def query_signals(
        self,
        strategy_id: str = None,
        symbol: str = None,
        start_date: str = None,
        end_date: str = None,
        limit: int = 200,
    ) -> dict:
        """查询信号"""
        payload = {"limit": limit}
        if strategy_id:
            payload["strategy_id"] = strategy_id
        if symbol:
            payload["symbol"] = symbol
        if start_date:
            payload["start_date"] = start_date
        if end_date:
            payload["end_date"] = end_date
        resp = self._client.post(f"{self.base_url}/signals/query", json=payload, headers=self._headers())
        resp.raise_for_status()
        return resp.json()

    # ═══════════════ 服务器端执行回测 ═══════════════

    def run_server_backtest(
        self,
        script_content: str,
        strategy_name: str,
        symbol: str,
        timeframe: str,
        start_date: str,
        end_date: str,
        strategy_id: str = "",
        initial_capital: float = 100_000.0,
        leverage: int = 1,
        fee_rate: float = 0.0005,
        slippage_bps: float = 5.0,
        margin_mode: str = "isolated",
        direction: str = "long_short",
    ) -> dict:
        """
        服务器端一站式回测 — 上传脚本，服务器执行+回测。

        与 run_backtest() 的区别:
        - run_backtest(): 本地跑脚本生成信号，只传信号给服务器
        - run_server_backtest(): 把脚本源码传给服务器，服务器执行一切
        """
        payload = {
            "script_content": script_content,
            "strategy_name": strategy_name,
            "strategy_id": strategy_id,
            "symbol": symbol,
            "timeframe": timeframe,
            "start_date": start_date,
            "end_date": end_date,
            "initial_capital": initial_capital,
            "leverage": leverage,
            "fee_rate": fee_rate,
            "slippage_bps": slippage_bps,
            "margin_mode": margin_mode,
            "direction": direction,
        }

        logger.info(
            "上传脚本到服务器执行 | {} {} {} | {} → {}",
            strategy_name, symbol, timeframe, start_date, end_date,
        )
        resp = self._client.post(
            f"{self.base_url}/backtest/run-server",
            json=payload,
            headers=self._headers(),
        )
        resp.raise_for_status()
        result = resp.json()

        status = result.get("status", "unknown")
        if status == "completed":
            metrics = result.get("metrics", {})
            logger.info(
                "服务器回测完成 | 收益={:.2%} | Sharpe={:.2f} | 交易={} | 结论={}",
                metrics.get("total_return_pct", 0),
                metrics.get("sharpe_ratio", 0),
                metrics.get("total_trades", 0),
                result.get("conclusion", ""),
            )
        else:
            logger.error("服务器回测失败 | {}", result.get("error"))

        return result

    # ═══════════════ 配额 ═══════════════

    def check_quota(self) -> dict:
        """查询当前机器码的策略配额"""
        return self._auth.check_quota()

    def print_quota(self) -> None:
        """打印配额信息"""
        self._auth.print_quota()

    # ═══════════════ 展示工具 ═══════════════

    @staticmethod
    def print_metrics(result: dict) -> None:
        """格式化打印回测结果"""
        if result.get("status") != "completed":
            print(f"回测失败: {result.get('error', '未知错误')}")
            return

        m = result.get("metrics", {})
        conclusion = result.get("conclusion", "")
        conclusion_map = {
            "approved": "✅ 通过 — 可以上线",
            "paper_trade_first": "⚠️  先模拟 — 建议观察",
            "rejected": "❌ 驳回 — 需要调整",
        }

        print("\n" + "=" * 55)
        print("  回测绩效报告")
        print("=" * 55)
        print(f"  策略:          {result.get('strategy_name', '')}")
        print(f"  结论:          {conclusion_map.get(conclusion, conclusion)}")
        print("-" * 55)
        print(f"  总收益率:      {m.get('total_return_pct', 0):>+10.2%}")
        print(f"  年化收益:      {m.get('annual_return_pct', 0):>+10.2%}")
        print(f"  Sharpe:        {m.get('sharpe_ratio', 0):>10.3f}")
        print(f"  Sortino:       {m.get('sortino_ratio', 0):>10.3f}")
        print(f"  最大回撤:      {abs(m.get('max_drawdown_pct', 0)):>10.2%}")
        print(f"  Calmar:        {m.get('calmar_ratio', 0):>10.3f}")
        print("-" * 55)
        print(f"  胜率:          {m.get('win_rate', 0):>10.2%}")
        print(f"  盈亏比:        {m.get('profit_loss_ratio', 0):>10.2f}")
        print(f"  总交易数:      {m.get('total_trades', 0):>10d}")
        print(f"  盈利交易:      {m.get('winning_trades', 0):>10d}")
        print(f"  亏损交易:      {m.get('losing_trades', 0):>10d}")
        print(f"  平均持仓:      {m.get('avg_holding_bars', 0):>10.1f} bars")
        print("-" * 55)
        print(f"  总手续费:      {m.get('total_commission', 0):>10.2f}")
        print(f"  总滑点成本:    {m.get('total_slippage_cost', 0):>10.2f}")
        print(f"  资金费率净值:  {m.get('net_funding', 0):>+10.2f}")
        print(f"  爆仓次数:      {m.get('liquidation_count', 0):>10d}")
        print("-" * 55)
        print(f"  总信号数:      {m.get('total_signals', 0):>10d}")
        print(f"  已执行信号:    {m.get('signals_executed', 0):>10d}")
        print(f"  最终余额:      {m.get('final_balance', 0):>10.2f}")
        print("=" * 55)

    @staticmethod
    def print_trades(result: dict, limit: int = 20) -> None:
        """格式化打印交易记录"""
        trades = result.get("trades", [])
        if not trades:
            print("无交易记录")
            return

        print(f"\n交易记录（共 {len(trades)} 笔，显示前 {min(limit, len(trades))} 笔）")
        print("-" * 100)
        print(f"{'#':>4} {'时间':<20} {'动作':<10} {'方向':<6} {'价格':>12} {'数量':>10} {'盈亏':>12} {'余额':>14}")
        print("-" * 100)

        for t in trades[:limit]:
            print(
                f"{t.get('trade_id', 0):>4} "
                f"{t.get('datetime', ''):<20} "
                f"{t.get('action', ''):<10} "
                f"{t.get('side', ''):<6} "
                f"{t.get('price', 0):>12.2f} "
                f"{t.get('quantity', 0):>10.4f} "
                f"{t.get('pnl', 0):>+12.2f} "
                f"{t.get('balance_after', 0):>14.2f}"
            )

        if len(trades) > limit:
            print(f"  ... 还有 {len(trades) - limit} 笔交易")
        print("-" * 100)

    @staticmethod
    def print_conclusion(result: dict) -> None:
        """打印回测结论和建议"""
        conclusion = result.get("conclusion", "")
        metrics = result.get("metrics", {})

        print("\n" + "=" * 50)
        if conclusion == "approved":
            print("  结论: 通过")
            print("  建议: 可以进入监控执行阶段")
        elif conclusion == "paper_trade_first":
            print("  结论: 先模拟")
            print("  建议: 先跑模拟盘观察 1-2 周")
        elif conclusion == "rejected":
            print("  结论: 驳回")
            print("  建议: 调整策略参数后重新回测")
        else:
            print(f"  结论: {conclusion}")

        if metrics:
            issues = []
            if metrics.get("total_return_pct", 0) < 0:
                issues.append("总收益为负")
            if abs(metrics.get("max_drawdown_pct", 0)) > 0.2:
                issues.append("最大回撤超过 20%")
            if metrics.get("sharpe_ratio", 0) < 1.0:
                issues.append("夏普比率低于 1.0")
            if metrics.get("win_rate", 0) < 0.3:
                issues.append("胜率低于 30%")
            if metrics.get("liquidation_count", 0) > 0:
                issues.append(f"发生 {metrics['liquidation_count']} 次爆仓")

            if issues:
                print("  风险点:")
                for issue in issues:
                    print(f"    - {issue}")
        print("=" * 50)

    # ═══════════════ 生命周期 ═══════════════

    def close(self):
        self._client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

"""
本地回测引擎 — 永续合约专项

支持:
  - 多空双向持仓
  - 逐仓/全仓保证金
  - 杠杆 1x-125x
  - 资金费率每 8h 结算（使用真实历史数据）
  - 强制平仓
  - 止损/止盈
  - 固定滑点
  - 手续费（Maker/Taker）
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional

import numpy as np
import pandas as pd
from loguru import logger


# ═══════════════════════════════════════════
#  数据结构
# ═══════════════════════════════════════════

@dataclass
class Position:
    """单个持仓。"""
    symbol: str
    side: str = "none"          # "long" / "short" / "none"
    quantity: float = 0.0
    avg_entry_price: float = 0.0
    leverage: int = 1
    margin: float = 0.0
    margin_mode: str = "isolated"
    maintenance_margin_rate: float = 0.005
    stop_loss: Optional[float] = None
    take_profit: Optional[float] = None
    unrealized_pnl: float = 0.0
    liquidation_price: float = 0.0

    @property
    def nominal_value(self) -> float:
        return self.quantity * self.avg_entry_price

    def calc_unrealized_pnl(self, mark_price: float) -> float:
        if self.side == "long":
            return self.quantity * (mark_price - self.avg_entry_price)
        elif self.side == "short":
            return self.quantity * (self.avg_entry_price - mark_price)
        return 0.0

    def calc_liquidation_price(self) -> float:
        """
        逐仓强平价格:
            多单: entry × (1 - 1/leverage + mmr)
            空单: entry × (1 + 1/leverage - mmr)
        """
        if self.quantity == 0 or self.side == "none":
            return 0.0
        mmr = self.maintenance_margin_rate
        if self.side == "long":
            return self.avg_entry_price * (1 - 1 / self.leverage + mmr)
        else:
            return self.avg_entry_price * (1 + 1 / self.leverage - mmr)

    def calc_margin_ratio(self, mark_price: float) -> float:
        """保证金率 = (保证金 + 未实现盈亏) / 名义价值"""
        nominal = self.quantity * mark_price
        if nominal == 0:
            return float("inf")
        pnl = self.calc_unrealized_pnl(mark_price)
        return (self.margin + pnl) / nominal


@dataclass
class TradeRecord:
    """单笔交易记录。"""
    datetime: str
    symbol: str
    side: str
    action: str           # "open" / "close" / "liquidation"
    quantity: float
    price: float
    mark_price: float
    leverage: int
    margin_used: float
    commission: float
    slippage: float
    funding_fee: float
    realized_pnl: float


@dataclass
class Account:
    """账户状态。"""
    initial_capital: float
    balance: float = 0.0
    positions: dict = field(default_factory=dict)
    trade_log: list = field(default_factory=list)
    equity_curve: list = field(default_factory=list)
    funding_log: list = field(default_factory=list)
    liquidation_count: int = 0
    total_commission: float = 0.0
    total_slippage_cost: float = 0.0
    total_funding_paid: float = 0.0
    total_funding_received: float = 0.0

    def __post_init__(self):
        self.balance = self.initial_capital

    def get_position(self, symbol: str) -> Position:
        if symbol not in self.positions:
            self.positions[symbol] = Position(symbol=symbol)
        return self.positions[symbol]

    @property
    def total_unrealized_pnl(self) -> float:
        return sum(p.unrealized_pnl for p in self.positions.values())

    @property
    def equity(self) -> float:
        return self.balance + self.total_unrealized_pnl

    @property
    def used_margin(self) -> float:
        return sum(p.margin for p in self.positions.values() if p.side != "none")

    @property
    def available_balance(self) -> float:
        return self.balance - self.used_margin


# ═══════════════════════════════════════════
#  回测配置
# ═══════════════════════════════════════════

@dataclass
class BacktestConfig:
    """回测引擎配置。所有参数均可外部化。"""
    initial_capital: float = 100_000.0
    default_leverage: int = 1
    margin_mode: str = "isolated"       # "isolated" / "cross"
    slippage_bps: float = 5.0           # 滑点（基点）
    taker_fee: float = 0.0005           # Taker 手续费 0.05%
    maker_fee: float = 0.0002           # Maker 手续费 0.02%
    enable_funding: bool = True         # 是否启用资金费率结算
    enable_liquidation: bool = True     # 是否启用强平检查
    maintenance_margin_rate: float = 0.005  # 维持保证金率 0.5%


# ═══════════════════════════════════════════
#  回测引擎
# ═══════════════════════════════════════════

class BacktestEngine:
    """
    永续合约回测引擎。

    使用方式:
        config = BacktestConfig(initial_capital=100000, default_leverage=5)
        engine = BacktestEngine(config)

        for i, row in df.iterrows():
            # 交易逻辑
            if signal_long:
                engine.open_long(symbol, qty, row["close"], row["close"], dt)
            # 每 bar 更新
            engine.on_bar(dt, prices, funding_rates)

        result = engine.get_result()
    """

    def __init__(self, config: BacktestConfig = None):
        self.config = config or BacktestConfig()
        self.account = Account(initial_capital=self.config.initial_capital)

    # ── 交易操作 ──

    def open_long(self, symbol: str, qty: float, price: float,
                  mark_price: float, dt: str, leverage: int = None):
        """开多仓。"""
        self._open_position(symbol, "long", qty, price, mark_price, dt, leverage)

    def open_short(self, symbol: str, qty: float, price: float,
                   mark_price: float, dt: str, leverage: int = None):
        """开空仓。"""
        self._open_position(symbol, "short", qty, price, mark_price, dt, leverage)

    def close_long(self, symbol: str, qty: float, price: float,
                   mark_price: float, dt: str):
        """平多仓。qty=0 表示全部平仓。"""
        self._close_position(symbol, "long", qty, price, mark_price, dt)

    def close_short(self, symbol: str, qty: float, price: float,
                    mark_price: float, dt: str):
        """平空仓。qty=0 表示全部平仓。"""
        self._close_position(symbol, "short", qty, price, mark_price, dt)

    def set_stop_loss(self, symbol: str, price: float):
        """设置止损价。"""
        self.account.get_position(symbol).stop_loss = price

    def set_take_profit(self, symbol: str, price: float):
        """设置止盈价。"""
        self.account.get_position(symbol).take_profit = price

    def set_leverage(self, symbol: str, leverage: int):
        """设置杠杆倍数 (1-125)。"""
        pos = self.account.get_position(symbol)
        pos.leverage = max(1, min(125, leverage))

    def set_margin_mode(self, symbol: str, mode: str):
        """设置保证金模式: 'isolated' 或 'cross'。"""
        pos = self.account.get_position(symbol)
        pos.margin_mode = mode

    def get_position(self, symbol: str) -> dict:
        """获取当前持仓信息。"""
        pos = self.account.get_position(symbol)
        return {
            "side": pos.side,
            "quantity": pos.quantity,
            "avg_entry_price": pos.avg_entry_price,
            "unrealized_pnl": pos.unrealized_pnl,
            "liquidation_price": pos.liquidation_price,
            "leverage": pos.leverage,
            "margin": pos.margin,
            "margin_ratio": pos.calc_margin_ratio(pos.avg_entry_price) if pos.side != "none" else 0,
        }

    # ── 内部: 开仓 ──

    def _open_position(self, symbol: str, side: str, qty: float, price: float,
                       mark_price: float, dt: str, leverage: int = None):
        pos = self.account.get_position(symbol)
        lev = leverage or pos.leverage or self.config.default_leverage
        pos.leverage = lev
        pos.margin_mode = pos.margin_mode or self.config.margin_mode
        pos.maintenance_margin_rate = self.config.maintenance_margin_rate

        slippage = price * self.config.slippage_bps / 10000
        fill_price = price + slippage if side == "long" else price - slippage

        nominal = qty * fill_price
        required_margin = nominal / lev
        commission = nominal * self.config.taker_fee

        if self.account.available_balance < required_margin + commission:
            logger.warning(
                f"[{dt}] 余额不足: 需要 {required_margin + commission:.2f}, "
                f"可用 {self.account.available_balance:.2f}"
            )
            return

        if pos.side == side and pos.quantity > 0:
            total_qty = pos.quantity + qty
            pos.avg_entry_price = (
                pos.avg_entry_price * pos.quantity + fill_price * qty
            ) / total_qty
            pos.quantity = total_qty
            pos.margin += required_margin
        else:
            pos.side = side
            pos.quantity = qty
            pos.avg_entry_price = fill_price
            pos.margin = required_margin

        pos.liquidation_price = pos.calc_liquidation_price()

        self.account.balance -= commission
        self.account.total_commission += commission
        self.account.total_slippage_cost += abs(slippage * qty)

        self.account.trade_log.append(TradeRecord(
            datetime=dt, symbol=symbol, side=side, action="open",
            quantity=qty, price=fill_price, mark_price=mark_price,
            leverage=lev, margin_used=required_margin,
            commission=commission, slippage=abs(slippage * qty),
            funding_fee=0.0, realized_pnl=0.0,
        ))

    # ── 内部: 平仓 ──

    def _close_position(self, symbol: str, side: str, qty: float, price: float,
                        mark_price: float, dt: str, action: str = "close"):
        pos = self.account.get_position(symbol)
        if pos.side != side or pos.quantity == 0:
            return

        close_qty = min(qty, pos.quantity) if qty else pos.quantity

        slippage = price * self.config.slippage_bps / 10000
        fill_price = price - slippage if side == "long" else price + slippage

        if side == "long":
            realized_pnl = close_qty * (fill_price - pos.avg_entry_price)
        else:
            realized_pnl = close_qty * (pos.avg_entry_price - fill_price)

        nominal = close_qty * fill_price
        commission = nominal * self.config.taker_fee

        margin_released = pos.margin * (close_qty / pos.quantity)
        pos.margin -= margin_released
        self.account.balance += margin_released + realized_pnl - commission
        self.account.total_commission += commission
        self.account.total_slippage_cost += abs(slippage * close_qty)

        pos.quantity -= close_qty
        if pos.quantity <= 1e-10:
            pos.quantity = 0
            pos.side = "none"
            pos.margin = 0
            pos.stop_loss = None
            pos.take_profit = None

        self.account.trade_log.append(TradeRecord(
            datetime=dt, symbol=symbol, side=side, action=action,
            quantity=close_qty, price=fill_price, mark_price=mark_price,
            leverage=pos.leverage, margin_used=0,
            commission=commission, slippage=abs(slippage * close_qty),
            funding_fee=0.0, realized_pnl=realized_pnl,
        ))

    # ── 每 bar 检查 ──

    def on_bar(self, dt: str, prices: dict[str, dict],
               funding_rates: dict[str, float] = None):
        """
        每个 bar 调用一次。

        执行顺序: 更新盈亏 → 资金费率结算 → 止损止盈 → 强平检查 → 记录净值

        参数:
            dt: 当前 bar 时间
            prices: {symbol: {"close": float, "high": float, "low": float, "mark_price": float}}
            funding_rates: {symbol: float} — 仅在 8h 结算时刻传入
        """
        for symbol, pos in list(self.account.positions.items()):
            if pos.side == "none":
                continue

            bar = prices.get(symbol, {})
            mark = bar.get("mark_price", bar.get("close", pos.avg_entry_price))

            pos.unrealized_pnl = pos.calc_unrealized_pnl(mark)

            if self.config.enable_funding and funding_rates and symbol in funding_rates:
                self._settle_funding(pos, funding_rates[symbol], mark, dt)

            if pos.side != "none":
                self._check_stop_loss_take_profit(pos, bar, dt)

            if self.config.enable_liquidation and pos.side != "none":
                self._check_liquidation(pos, mark, dt)

        self.account.equity_curve.append({
            "datetime": dt,
            "equity": self.account.equity,
            "balance": self.account.balance,
            "unrealized_pnl": self.account.total_unrealized_pnl,
            "used_margin": self.account.used_margin,
            "drawdown": 0.0,
        })

    def _settle_funding(self, pos: Position, funding_rate: float,
                        mark_price: float, dt: str):
        """资金费率结算。"""
        nominal = pos.quantity * mark_price
        fee = nominal * funding_rate

        if pos.side == "long":
            pos.margin -= fee
            self.account.balance -= fee
            if fee > 0:
                self.account.total_funding_paid += fee
            else:
                self.account.total_funding_received += abs(fee)
        else:
            pos.margin += fee
            self.account.balance += fee
            if fee > 0:
                self.account.total_funding_received += fee
            else:
                self.account.total_funding_paid += abs(fee)

        self.account.funding_log.append({
            "datetime": dt,
            "symbol": pos.symbol,
            "side": pos.side,
            "funding_rate": funding_rate,
            "position_value": nominal,
            "fee": fee,
        })

    def _check_stop_loss_take_profit(self, pos: Position, bar: dict, dt: str):
        """检查止损/止盈是否触发。"""
        high = bar.get("high", bar.get("close", 0))
        low = bar.get("low", bar.get("close", 0))
        mark = bar.get("mark_price", bar.get("close", 0))

        if pos.stop_loss is not None:
            triggered = (
                (pos.side == "long" and low <= pos.stop_loss)
                or (pos.side == "short" and high >= pos.stop_loss)
            )
            if triggered:
                logger.info(f"[{dt}] 止损触发: {pos.symbol} {pos.side} @ {pos.stop_loss}")
                self._close_position(
                    pos.symbol, pos.side, pos.quantity,
                    pos.stop_loss, mark, dt, "close",
                )
                return

        if pos.take_profit is not None:
            triggered = (
                (pos.side == "long" and high >= pos.take_profit)
                or (pos.side == "short" and low <= pos.take_profit)
            )
            if triggered:
                logger.info(f"[{dt}] 止盈触发: {pos.symbol} {pos.side} @ {pos.take_profit}")
                self._close_position(
                    pos.symbol, pos.side, pos.quantity,
                    pos.take_profit, mark, dt, "close",
                )

    def _check_liquidation(self, pos: Position, mark_price: float, dt: str):
        """强平检查。"""
        if pos.side == "none" or pos.quantity == 0:
            return

        margin_ratio = pos.calc_margin_ratio(mark_price)

        if margin_ratio <= pos.maintenance_margin_rate:
            logger.warning(
                f"[{dt}] 强平: {pos.symbol} {pos.side} "
                f"保证金率 {margin_ratio:.4f} <= {pos.maintenance_margin_rate}"
            )
            lost_margin = pos.margin
            pos.quantity = 0
            pos.side = "none"
            pos.margin = 0
            pos.unrealized_pnl = 0
            pos.stop_loss = None
            pos.take_profit = None

            self.account.liquidation_count += 1
            self.account.trade_log.append(TradeRecord(
                datetime=dt, symbol=pos.symbol, side=pos.side, action="liquidation",
                quantity=0, price=mark_price, mark_price=mark_price,
                leverage=pos.leverage, margin_used=0,
                commission=0, slippage=0, funding_fee=0,
                realized_pnl=-lost_margin,
            ))

    # ── 结果汇总 ──

    def get_result(self) -> dict:
        """获取完整回测结果。"""
        eq_df = pd.DataFrame(self.account.equity_curve)
        if eq_df.empty:
            return {"error": "无回测数据"}

        equities = eq_df["equity"].values
        peak = np.maximum.accumulate(equities)
        drawdowns = (equities - peak) / peak
        eq_df["drawdown"] = drawdowns

        returns = np.diff(equities) / equities[:-1] if len(equities) > 1 else np.array([0])

        total_return = (equities[-1] / equities[0]) - 1
        n_days = len(equities)
        annual_return = (1 + total_return) ** (365 / max(n_days, 1)) - 1
        volatility = float(np.std(returns) * np.sqrt(365)) if len(returns) > 1 else 0

        rf = 0.0
        sharpe = (annual_return - rf) / volatility if volatility > 0 else 0
        downside = returns[returns < 0]
        downside_std = float(np.std(downside) * np.sqrt(365)) if len(downside) > 0 else 0
        sortino = (annual_return - rf) / downside_std if downside_std > 0 else 0

        max_dd = float(np.min(drawdowns))
        max_dd_idx = int(np.argmin(drawdowns))
        peak_idx = int(np.argmax(equities[:max_dd_idx + 1])) if max_dd_idx > 0 else 0
        max_dd_duration = max_dd_idx - peak_idx

        calmar = annual_return / abs(max_dd) if max_dd != 0 else 0

        trades = self.account.trade_log
        close_trades = [
            t for t in trades
            if t.action in ("close", "liquidation") and t.realized_pnl != 0
        ]
        wins = [t for t in close_trades if t.realized_pnl > 0]
        losses = [t for t in close_trades if t.realized_pnl < 0]
        win_rate = len(wins) / len(close_trades) if close_trades else 0
        avg_win = float(np.mean([t.realized_pnl for t in wins])) if wins else 0
        avg_loss = float(abs(np.mean([t.realized_pnl for t in losses]))) if losses else 0
        profit_loss_ratio = avg_win / avg_loss if avg_loss > 0 else float("inf")

        net_funding = self.account.total_funding_received - self.account.total_funding_paid

        return {
            "performance": {
                "total_return": total_return,
                "annual_return": annual_return,
                "sharpe_ratio": sharpe,
                "sortino_ratio": sortino,
                "max_drawdown": max_dd,
                "max_drawdown_duration": int(max_dd_duration),
                "calmar_ratio": calmar,
                "volatility": volatility,
                "win_rate": win_rate,
                "profit_loss_ratio": profit_loss_ratio,
                "total_trades": len(trades),
                "total_funding_paid": self.account.total_funding_paid,
                "total_funding_received": self.account.total_funding_received,
                "net_funding": net_funding,
                "total_commission": self.account.total_commission,
                "total_slippage_cost": self.account.total_slippage_cost,
                "liquidation_count": self.account.liquidation_count,
            },
            "equity_curve": eq_df.to_dict("records"),
            "trade_log": [vars(t) for t in trades],
            "funding_log": self.account.funding_log,
        }

    def get_metrics(self) -> dict:
        """获取绩效指标摘要（不含完整曲线和日志）。"""
        result = self.get_result()
        return result.get("performance", result)

    @staticmethod
    def format_summary(result: dict) -> str:
        """格式化输出回测结果摘要。"""
        p = result.get("performance", {})
        lines = [
            "═══ 回测结果摘要 ═══",
            f"总收益率:     {p.get('total_return', 0):.2%}",
            f"年化收益率:   {p.get('annual_return', 0):.2%}",
            f"夏普比率:     {p.get('sharpe_ratio', 0):.3f}",
            f"索提诺比率:   {p.get('sortino_ratio', 0):.3f}",
            f"最大回撤:     {p.get('max_drawdown', 0):.2%}",
            f"卡尔玛比率:   {p.get('calmar_ratio', 0):.3f}",
            f"年化波动率:   {p.get('volatility', 0):.2%}",
            f"胜率:         {p.get('win_rate', 0):.2%}",
            f"盈亏比:       {p.get('profit_loss_ratio', 0):.2f}",
            f"总交易次数:   {p.get('total_trades', 0)}",
            "─── 资金费率 ───",
            f"支付:         {p.get('total_funding_paid', 0):.2f} USDT",
            f"收到:         {p.get('total_funding_received', 0):.2f} USDT",
            f"净损益:       {p.get('net_funding', 0):.2f} USDT",
            "─── 费用与风控 ───",
            f"累计手续费:   {p.get('total_commission', 0):.2f} USDT",
            f"累计滑点:     {p.get('total_slippage_cost', 0):.2f} USDT",
            f"强平次数:     {p.get('liquidation_count', 0)}",
        ]
        return "\n".join(lines)