Vibe Trading

root = true

[*]
charset = utf-8
end_of_line = lf
insert_final_newline = true
trim_trailing_whitespace = true

# ============================================================================
# LLM Provider — uncomment ONE provider block below
# ============================================================================

# --- OpenRouter (recommended, multi-model gateway) ---
LANGCHAIN_PROVIDER=openrouter
LANGCHAIN_MODEL_NAME=deepseek/deepseek-v4-pro
OPENROUTER_API_KEY=sk-or-v1-your-key-here
OPENROUTER_BASE_URL=https://openrouter.ai/api/v1

# --- OpenAI ---
# LANGCHAIN_PROVIDER=openai
# LANGCHAIN_MODEL_NAME=gpt-5.5-instant
# OPENAI_API_KEY=sk-xxx
# OPENAI_BASE_URL=https://api.openai.com/v1

# --- OpenAI Codex (ChatGPT OAuth; not the OpenAI API key path) ---
# Login first: vibe-trading provider login openai-codex
# Requires a ChatGPT account with Codex access. OAuth tokens are stored by oauth-cli-kit.
# LANGCHAIN_PROVIDER=openai-codex
# LANGCHAIN_MODEL_NAME=openai-codex/gpt-5.3-codex
# OPENAI_CODEX_BASE_URL=https://chatgpt.com/backend-api/codex/responses

# --- DeepSeek ---
# LANGCHAIN_PROVIDER=deepseek
# LANGCHAIN_MODEL_NAME=deepseek-v4-pro
# DEEPSEEK_API_KEY=sk-xxx
# DEEPSEEK_BASE_URL=https://api.deepseek.com/v1
# DeepSeek adapter selection: auto (default; use langchain-deepseek if installed),
# native (require langchain-deepseek, error if missing), or openai-compatible
# (force the legacy ChatOpenAI path). Install the adapter with:
# pip install "vibe-trading-ai[deepseek]"
# VIBE_TRADING_DEEPSEEK_ADAPTER=auto

# --- Gemini ---
# LANGCHAIN_PROVIDER=gemini
# LANGCHAIN_MODEL_NAME=gemini-3.5-flash
# GEMINI_API_KEY=xxx
# GEMINI_BASE_URL=https://generativelanguage.googleapis.com/v1beta/openai/

# --- Groq ---
# LANGCHAIN_PROVIDER=groq
# LANGCHAIN_MODEL_NAME=meta-llama/llama-4-maverick-17b-128e-instruct
# GROQ_API_KEY=gsk_xxx
# GROQ_BASE_URL=https://api.groq.com/openai/v1

# --- DashScope / Qwen ---
# LANGCHAIN_PROVIDER=dashscope
# LANGCHAIN_MODEL_NAME=qwen-plus-latest
# DASHSCOPE_API_KEY=sk-xxx
# DASHSCOPE_BASE_URL=https://dashscope.aliyuncs.com/compatible-mode/v1

# --- Zhipu ---
# LANGCHAIN_PROVIDER=zhipu
# LANGCHAIN_MODEL_NAME=glm-5.1
# ZHIPU_API_KEY=xxx
# ZHIPU_BASE_URL=https://open.bigmodel.cn/api/paas/v4

# --- Moonshot / Kimi ---
# LANGCHAIN_PROVIDER=moonshot
# LANGCHAIN_MODEL_NAME=kimi-k2.6
# MOONSHOT_API_KEY=sk-xxx
# MOONSHOT_BASE_URL=https://api.moonshot.ai/v1

# --- MiniMax ---
# LANGCHAIN_PROVIDER=minimax
# LANGCHAIN_MODEL_NAME=MiniMax-M3              # or MiniMax-M2.7 / MiniMax-M2.7-highspeed
# MINIMAX_API_KEY=xxx
# MINIMAX_BASE_URL=https://api.minimax.io/v1
# Note: MiniMax requires temperature > 0. Set LANGCHAIN_TEMPERATURE=1.0 (default when using MiniMax)

# --- Xiaomi MIMO ---
# LANGCHAIN_PROVIDER=mimo
# LANGCHAIN_MODEL_NAME=MiMo-72B-A27B
# MIMO_API_KEY=xxx
# MIMO_BASE_URL=https://api.xiaomimimo.com/v1

# --- Z.ai (for Coding Plan) ---
# LANGCHAIN_PROVIDER=zai
# LANGCHAIN_MODEL_NAME=glm-5.1
# ZAI_API_KEY=xxx
# ZAI_BASE_URL=https://api.z.ai/api/coding/paas/v4

# --- Ollama (local) ---
# LANGCHAIN_PROVIDER=ollama
# LANGCHAIN_MODEL_NAME=qwen2.5:32b
# OLLAMA_BASE_URL=http://localhost:11434

# LLM parameters
LANGCHAIN_TEMPERATURE=0.0
TIMEOUT_SECONDS=120
MAX_RETRIES=2

# Enable reasoning on relays that require opt-in (e.g. OpenRouter thinking models).
# Forwarded as extra_body.reasoning.effort = low / medium / high / max.
# Leave unset for Moonshot / DeepSeek official APIs — they return reasoning_content by default.
# LANGCHAIN_REASONING_EFFORT=medium

# ============================================================================
# Data Sources
# ============================================================================
# A-shares: Tushare Pro token (https://tushare.pro)
TUSHARE_TOKEN=your-tushare-token
# HK/US equities: yfinance (free, no config needed)
# Crypto: OKX public API (free, no config needed)
# Crypto fallback exchange (default: binance). Change if OKX is blocked:
# CCXT_EXCHANGE=binance
# HK / A-share equities via Futu OpenAPI (optional, requires FutuOpenD running locally)
# Download FutuOpenD: https://www.futunn.com/download/openAPI
# FUTU_HOST=127.0.0.1
# FUTU_PORT=11111
# Free direct-API sources need NO key and auto-join the fallback chain: Eastmoney / Sina / Stooq / Yahoo.

# Optional API-key data sources (enabled only when the key is set; silently skipped otherwise)
# FINNHUB_API_KEY=xxx         # optional US-equity OHLCV fallback provider (selectable get_market_data source)
# ALPHAVANTAGE_API_KEY=xxx    # optional US-equity OHLCV fallback provider (selectable get_market_data source)
# TIINGO_API_KEY=xxx          # optional US-equity OHLCV fallback provider (selectable get_market_data source)
# FMP_API_KEY=xxx             # optional US-equity OHLCV fallback provider (selectable get_market_data source)
# FRED_API_KEY=xxx            # macro series (get_macro_series tool)
# VIBE_TRADING_IWENCAI_KEY=xxx  # A-share natural-language research search (iwencai_search tool)
# VIBE_TRADING_SEC_UA="Your Name your@email.com"  # SEC EDGAR compliant contact User-Agent (optional; a built-in default UA also works)
# Optional per-source request-spacing overrides for batch jobs — VIBE_TRADING_{EASTMONEY,SINA,STOOQ,YAHOO,SEC,FINNHUB,ALPHAVANTAGE,TIINGO,FMP,FRED,IWENCAI,THS}_MIN_INTERVAL (seconds).

# Opt-in local market-data cache for backtest loaders (off by default).
# When enabled, every data source caches settled historical bars under
# ~/.vibe-trading/cache/loaders/ so repeated/long-horizon backtests skip the
# network. Only fully-elapsed days are cached (ranges ending today are always
# re-fetched). Clear it any time with: rm -rf ~/.vibe-trading/cache
# VIBE_TRADING_DATA_CACHE=1

# ============================================================================
# API Server (optional)
# ============================================================================
# Bearer token for API auth.
# Empty dev mode is loopback-only. Set this before exposing port 8899 beyond localhost.
# API_AUTH_KEY=
# CORS allowed origins (comma-separated)
# CORS_ORIGINS=http://localhost:3000,http://localhost:5173,http://localhost:8000
# Enable session runtime
# ENABLE_SESSION_RUNTIME=true
# Docker compose sets this together with a 127.0.0.1 port bind so browser
# requests from the host machine still count as local dev-mode access.
# Do not enable this when publishing port 8899 on 0.0.0.0 without API_AUTH_KEY.
# VIBE_TRADING_TRUST_DOCKER_LOOPBACK=0
# Advanced local-only opt-in: expose shell execution tools to the agent.
# VIBE_TRADING_ENABLE_SHELL_TOOLS=0
# Optional import roots for read_document/analyze_trade_journal.
# Use comma-separated absolute paths. Defaults include agent/uploads, agent/runs,
# ./uploads, ./data, ~/.vibe-trading/uploads, and ~/.vibe-trading/imports.
# VIBE_TRADING_ALLOWED_FILE_ROOTS=
# Optional run roots for generated-code tools such as write_file/backtest.
# Defaults include agent/runs, ./runs, and ~/.vibe-trading/shadow_runs.
# VIBE_TRADING_ALLOWED_RUN_ROOTS=

# ============================================================================
# Agent Tuning (optional, defaults are sensible)
# ============================================================================
# SWARM_WORKER_TIMEOUT=300
# SWARM_WORKER_MAX_ITER=50
# SWARM_MAX_WORKERS=4
# SWARM_TIMEOUT=1800
# SUBAGENT_TIMEOUT=300
# SUBAGENT_MAX_ITER=25
# TOKEN_THRESHOLD=40000
# Hard timeout (seconds) for read-only tool execution; write tools only warn.
# Default 1800; set 0 to disable.
# VIBE_TRADING_TOOL_TIMEOUT_SECONDS=1800
# Frontend SSE idle timeout in seconds before showing "Execution timed out".
# Increase when using slow local models (e.g. Ollama on CPU/limited VRAM).
# VIBE_TRADING_SSE_TIMEOUT=90

.venv/
.ui_runtime/
.vscode/
__pycache__/
*.pyc

runs/

.env

"""Benchmark ticker resolution and fetch for backtest comparison.

Provides a lightweight, zero-dependency way to fetch benchmark reference
data given a set of strategy codes and a data source.
"""

from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

import pandas as pd

from backtest.loaders.yfinance_loader import DataLoader as YfinanceLoader


# -------------------------------------------------------------------
# Benchmark map: market type → default ticker
# -------------------------------------------------------------------

MARKET_BENCHMARKS: dict[str, Optional[str]] = {
    "us_equity":  "SPY",
    "hk_equity":  "HK.03100",   # Hang Seng China Enterprises ETF
    "a_share":    "000300.SH",  # CSI 300 (China A-share core index)
    "crypto":     "BTC-USDT",
    "futures":    "ES.CME",      # E-mini S&P 500 futures
    "forex":      None,         # no universal benchmark
}


@dataclass
class BenchmarkResult:
    ticker:     str
    ret_series: pd.Series       # per-bar returns, index = timestamps
    total_ret: float          # total return over the period


def resolve_benchmark(
    strategy_codes: list[str],
    source:       str,
    start_date:   str,
    end_date:     str,
    interval:     str = "1D",
    explicit:     Optional[str] = None,
) -> Optional[BenchmarkResult]:
    """Resolve the appropriate benchmark ticker and fetch its return series.

    Args:
        strategy_codes: Instruments being backtested (used for market inference).
        source:         Data source name (tushare / yfinance / okx / akshare / ccxt).
        start_date:     Backtest start date.
        end_date:       Backtest end date.
        interval:       Bar interval (1m / 5m / 15m / 30m / 1H / 4H / 1D).
        explicit:       Override ticker (e.g. "SPY" passed via config).

    Returns:
        BenchmarkResult with return series and total return, or None if no
        benchmark applies (forex, or fetch failure).
    """
    ticker = _resolve_ticker(strategy_codes, source, explicit)
    if ticker is None:
        return None

    try:
        bench_df = _fetch_benchmark(ticker, start_date, end_date, interval)
    except Exception:
        return None

    if bench_df.empty or "close" not in bench_df.columns:
        return None

    close = bench_df["close"].dropna()
    if len(close) < 2:
        return None

    ret_series = close.pct_change().fillna(0.0)
    total_ret   = float((1 + ret_series).prod() - 1)

    return BenchmarkResult(ticker=ticker, ret_series=ret_series, total_ret=total_ret)


# -------------------------------------------------------------------
# Internal helpers
# -------------------------------------------------------------------

def _resolve_ticker(
    codes:     list[str],
    source:    str,
    explicit:  Optional[str],
) -> Optional[str]:
    """Pick the benchmark ticker to use."""

    if explicit:
        return explicit

    # Infer market from source + first code pattern
    market = _infer_market(codes, source)
    ticker = MARKET_BENCHMARKS.get(market)

    # yfinance is the universal fallback for benchmark fetch
    # but it only works for us_equity / hk_equity market types
    if ticker and market not in {"us_equity", "hk_equity"}:
        # Only use benchmark if we can actually fetch it
        pass

    return ticker


def _infer_market(codes: list[str], source: str) -> str:
    """Rough market inference from symbol patterns and source."""
    if not codes:
        return "us_equity"

    first = codes[0].upper()

    if source in ("okx", "ccxt") or "-" in first or "/" in first:
        return "crypto"
    if first.endswith(".US"):
        return "us_equity"
    if first.endswith(".HK"):
        return "hk_equity"
    if source in ("tushare", "akshare"):
        if first.isdigit() and len(first) == 6:
            return "a_share"
        if first.startswith(("IF", "IC", "IH", "IM", "T", "TF")):
            return "futures"
        return "a_share"

    return "us_equity"


def _fetch_benchmark(
    ticker:    str,
    start_date: str,
    end_date:   str,
    interval:   str,
) -> pd.DataFrame:
    """Fetch benchmark OHLCV data via yfinance (single symbol, no auth)."""
    loader = YfinanceLoader()
    result = loader.fetch([ticker], start_date, end_date, interval=interval)

    if isinstance(result, dict):
        df = result.get(ticker)
    elif isinstance(result, pd.DataFrame):
        df = result
    else:
        return pd.DataFrame()

    if df is None or (isinstance(df, pd.DataFrame) and df.empty):
        return pd.DataFrame()

    return df

"""Cross-asset correlation matrix computation.

Computes pairwise Pearson or Spearman correlation of daily returns
over a configurable lookback window. Used by the /correlation API endpoint.
"""

from __future__ import annotations

from typing import Dict, Literal

import pandas as pd
import numpy as np
from scipy.stats import spearmanr


def infer_market(code: str) -> str:
    """Infer market key from a ticker symbol."""
    code_upper = code.upper()
    crypto_suffixes = ("USDT", "BTC", "ETH", "BNB", "SOL", "ADA", "DOGE")
    if any(code_upper.endswith(s) for s in crypto_suffixes) or "/" in code:
        return "crypto"
    # Check .HK suffix FIRST so leading-zero tickers like 0700.HK / 0005.HK
    # are correctly classified before the A-share prefix checks
    if code_upper.endswith(".HK"):
        return "hk_equity"
    if code_upper.startswith(("6", "000", "001", "002")):
        return "a_share"
    if code_upper.startswith(("0", "399")):
        return "a_share"
    if code_upper.startswith(("0", "1", "2", "3", "4")):
        return "hk_equity"
    return "us_equity"


def _rolling_correlation_matrix(
    price_series: Dict[str, pd.DataFrame],
    window: int,
    method: Literal["pearson", "spearman"],
) -> tuple[list[str], list[list[float]]]:
    """Compute correlation matrix for multiple price series.

    Args:
        price_series: Mapping of asset code -> DataFrame with a ``close`` column.
        window: Rolling window size in days.
        method: "pearson" or "spearman".

    Returns:
        (labels, matrix) where labels is the sorted list of codes and matrix
        is a symmetric NxN matrix of correlation coefficients.
    """
    if not price_series:
        return [], []

    codes = sorted(price_series.keys())

    # Build a aligned returns DataFrame (row index = date)
    returns_frames = []
    closes = {}
    for code, df in price_series.items():
        if df.empty:
            raise ValueError(f"Price series for '{code}' is empty")
        if "close" not in df.columns and "close" not in df.index.names:
            raise ValueError(f"No 'close' column in price series for '{code}'")
        # Support both column-based and index-based trade_date
        if "trade_date" in df.index.names and "trade_date" not in df.columns:
            ts = df["close"]
        else:
            ts = df.set_index("trade_date")["close"]
        closes[code] = ts.sort_index()

    for code in codes:
        ts = closes[code]
        # Normalize to date-only (midnight) so that cross-market assets
        # (e.g. crypto via OKX/CCXT at UTC midnight vs US equity via
        # yfinance at EDT midnight = 04:00 UTC) align correctly.
        ts.index = ts.index.normalize()
        rets = ts.pct_change().dropna()
        rets.name = code
        returns_frames.append(rets)

    # Align all series to a common index (inner join)
    aligned = pd.concat(returns_frames, axis=1).dropna()
    if aligned.empty:
        ranges = {
            code: f"{closes[code].index.min()} .. {closes[code].index.max()}"
            for code in codes
            if len(closes[code]) > 0
        }
        raise ValueError(
            f"No overlapping return data between assets. "
            f"Date ranges: {ranges}"
        )

    # Apply the trailing window — only use the last `window` rows of aligned data
    if len(aligned) > window:
        aligned = aligned.iloc[-window:]

    n = len(aligned)
    if n < 2:
        raise ValueError("Not enough data points to compute correlation")

    labels = codes
    n_assets = len(labels)
    matrix = [[1.0] * n_assets for _ in range(n_assets)]

    for i in range(n_assets):
        for j in range(i + 1, n_assets):
            xi = aligned.iloc[:, i].values
            xj = aligned.iloc[:, j].values
            if method == "spearman":
                corr, _ = spearmanr(xi, xj)
            else:
                corr = np.corrcoef(xi, xj)[0, 1]
            if np.isnan(corr):
                corr = 0.0
            matrix[i][j] = round(corr, 4)
            matrix[j][i] = round(corr, 4)

    return labels, matrix


def compute_correlation_matrix(
    codes: list[str],
    days: int = 90,
    method: Literal["pearson", "spearman"] = "pearson",
) -> Dict[str, object]:
    """Fetch price data and compute correlation matrix for a list of assets.

    Args:
        codes: List of asset codes (e.g. ["BTC-USDT", "ETH-USDT", "SPY"]).
        days: Lookback window in days (default 90).
        method: Correlation method.

    Returns:
        Dict with keys: labels, matrix, window, method.
    """
    from datetime import datetime, timedelta

    end_date = datetime.now().strftime("%Y-%m-%d")
    start_date = (datetime.now() - timedelta(days=days + 60)).strftime("%Y-%m-%d")

    # Import here to avoid circular
    from backtest.loaders.registry import resolve_loader

    price_series: Dict[str, pd.DataFrame] = {}

    for code in codes:
        market = infer_market(code)
        try:
            loader = resolve_loader(market)
        except Exception:
            # Fall back to yfinance for us_equity / hk_equity
            try:
                from backtest.loaders.registry import LOADER_REGISTRY
                if "yfinance" in LOADER_REGISTRY:
                    loader = LOADER_REGISTRY["yfinance"]()
                else:
                    continue
            except Exception:
                continue

        try:
            result = loader.fetch(
                codes=[code],
                start_date=start_date,
                end_date=end_date,
                interval="1D",
                fields=["trade_date", "open", "high", "low", "close", "volume"],
            )
            if code in result and not result[code].empty:
                price_series[code] = result[code]
        except Exception:
            continue

    if len(price_series) < 2:
        raise ValueError(
            f"Could not fetch price data for at least 2 assets. "
            f"Fetched: {list(price_series.keys())}"
        )

    labels, matrix = _rolling_correlation_matrix(price_series, days, method)
    return {
        "labels": labels,
        "matrix": matrix,
        "window": days,
        "method": method,
    }

"""Backtest engines.

Wave 1 (v1):
  - BaseEngine: ABC for bar-by-bar execution with market rules
  - ChinaAEngine: A-share (T+1, no short, price limits)
  - GlobalEquityEngine: US / HK equities
  - CryptoEngine: Crypto perpetuals (funding fees, liquidation)
  - options_portfolio: European/American options (Black-Scholes, v2 with IV smile)

Wave 2:
  - FuturesBaseEngine: intermediate layer adding contract-multiplier logic
  - ChinaFuturesEngine: China commodity/financial futures (CFFEX/SHFE/DCE/ZCE/INE)
  - GlobalFuturesEngine: International futures (CME/ICE/Eurex)
  - ForexEngine: FX spot/CFD (spread, swap, high leverage)

Wave 3:
  - CompositeEngine: Cross-market engine with shared capital pool
  - _market_hooks: Extracted on_bar logic (funding, liquidation, swap)

Inheritance:
  BaseEngine
  ├── ChinaAEngine
  ├── GlobalEquityEngine
  ├── CryptoEngine
  ├── ForexEngine
  ├── CompositeEngine (delegates to sub-engines as rule providers)
  └── FuturesBaseEngine
      ├── ChinaFuturesEngine
      └── GlobalFuturesEngine
"""

"""Extracted per-bar market hooks and symbol-classification helpers.

Both the original engines (CryptoEngine, ForexEngine) and CompositeEngine
call these same functions. Zero duplication — one source of truth.

Also hosts symbol -> market detection helpers shared by ``runner.py`` and
``composite.py``: ``_MARKET_PATTERNS``, ``_detect_market``,
``_is_china_futures``, ``_detect_submarket``. Keep regex tables here so the
truncated-duplicate routing bug (bare ``RB2410`` getting routed to
GlobalFutures because composite.py used a suffix-only check) cannot recur.
"""

from __future__ import annotations

import re
from typing import Dict, List

import pandas as pd

from backtest.models import Position


# ── Symbol -> market classification (shared by runner.py + composite.py) ──

_MARKET_PATTERNS = [
    (re.compile(r"^\d{6}\.(SZ|SH|BJ)$", re.I), "a_share"),
    (re.compile(r"^(51|15|56)\d{4}\.(SZ|SH)$", re.I), "a_share"),
    (re.compile(r"^[A-Z]+\.US$", re.I), "us_equity"),
    (re.compile(r"^\d{3,5}\.HK$", re.I), "hk_equity"),
    (re.compile(r"^[A-Z]+-USDT$", re.I), "crypto"),
    (re.compile(r"^[A-Z]+/USDT$", re.I), "crypto"),
    # China futures: product+delivery.exchange (e.g. IF2406.CFFEX, rb2410.SHFE)
    (re.compile(r"^[A-Za-z]{1,2}\d{3,4}\.(ZCE|DCE|SHFE|INE|CFFEX|GFEX)$", re.I), "futures"),
    # Global futures: product+month-code (e.g. ESZ4, CLF25, GCM2025)
    (re.compile(r"^[A-Z]{2,4}[FGHJKMNQUVXZ]\d{1,2}$", re.I), "futures"),
    # Global futures: product+YYMM (e.g. CL2412, ES2503)
    (re.compile(r"^[A-Z]{2,4}\d{4}$", re.I), "futures"),
    # Global futures: bare product code with exchange (e.g. ES.CME)
    (re.compile(r"^[A-Z]{2,4}\.(CME|CBOT|NYMEX|COMEX|ICE|EUREX)$", re.I), "futures"),
    # Forex pairs: XXX/YYY or XXXXXX.FX
    (re.compile(r"^[A-Z]{3}/[A-Z]{3}$"), "forex"),
    (re.compile(r"^[A-Z]{6}\.FX$"), "forex"),
]

_CHINA_EXCHANGES = {"CFFEX", "SHFE", "DCE", "ZCE", "INE", "GFEX"}

# Known Chinese-futures product codes — used as a heuristic when a symbol
# lacks an exchange suffix (e.g. bare ``RB2410``, ``IF2406``). Without this
# table composite.py was misrouting such bare codes to GlobalFutures.
# Stored lowercase; ``_is_china_futures`` lowercases the extracted product
# before lookup so callers can pass any case (``RB2410`` and ``rb2410``
# both resolve correctly).
_CN_FUTURES_PRODUCTS = {
    "if", "ic", "ih", "im", "t", "tf", "ts", "tl",
    "au", "ag", "cu", "al", "zn", "pb", "ni", "sn", "ss",
    "rb", "hc", "i", "j", "jm",
    "sc", "fu", "lu", "bu", "nr",
    "c", "cs", "m", "y", "a", "p", "jd", "lh",
    "cf", "sr", "ta", "ma", "ap", "rm", "oi",
    "pp", "l", "v", "eg", "eb", "pf", "sa", "fg", "ur",
    "si", "lc",
}


def _detect_market(code: str) -> str:
    """Infer market type from symbol format.

    Args:
        code: Ticker / symbol string.

    Returns:
        Market type (a_share/us_equity/hk_equity/crypto/futures/forex);
        unknown defaults to ``a_share``.
    """
    for pattern, market in _MARKET_PATTERNS:
        if pattern.match(code):
            return market
    return "a_share"


def _is_china_futures(code: str) -> bool:
    """Check whether a futures code belongs to a Chinese exchange.

    Recognises two forms:
      1. ``<product><delivery>.<exchange>`` where exchange is one of
         CFFEX/SHFE/DCE/ZCE/INE/GFEX (e.g. ``IF2406.CFFEX``, ``rb2410.SHFE``).
      2. Bare ``<product><delivery>`` with no exchange suffix — matched
         against ``_CN_FUTURES_PRODUCTS`` (e.g. ``RB2410`` -> True).

    Args:
        code: Symbol string.

    Returns:
        True if it looks like a Chinese futures contract.
    """
    parts = code.upper().split(".")
    if len(parts) == 2:
        # Has an exchange suffix — trust it. CN exchange = True, anything
        # else = False. Without this guard the product-code heuristic below
        # would misclassify global futures whose product letters happen to
        # collide with a CN product (e.g. ``M2412.CBOT`` — US soybean meal).
        return parts[1] in _CHINA_EXCHANGES
    # Bare code (no exchange suffix): fall back to product-code heuristic.
    m = re.match(r"([A-Za-z]+)\d+", parts[0])
    if m:
        product = m.group(1).lower()
        if product in _CN_FUTURES_PRODUCTS:
            return True
    return False


def _detect_submarket(codes: List[str]) -> str:
    """Detect US vs HK from symbol suffixes.

    Args:
        codes: Instrument codes.

    Returns:
        ``"hk"`` if any code ends with ``.HK``, else ``"us"``.
    """
    for code in codes:
        if code.upper().endswith(".HK"):
            return "hk"
    return "us"

# ── Crypto: OKX tiered maintenance margin table (simplified) ──

_TIER_TABLE = [
    (100_000, 0.004),
    (500_000, 0.006),
    (1_000_000, 0.01),
    (5_000_000, 0.02),
    (10_000_000, 0.05),
    (float("inf"), 0.10),
]

FUNDING_HOURS = {0, 8, 16}


def _maintenance_rate(notional_usd: float) -> float:
    """Look up tiered maintenance margin rate."""
    for tier_max, rate in _TIER_TABLE:
        if notional_usd <= tier_max:
            return rate
    return _TIER_TABLE[-1][1]


def calc_crypto_funding_fee(
    symbol: str,
    bar: pd.Series,
    timestamp: pd.Timestamp,
    positions: Dict[str, Position],
    funding_rate: float,
    applied_set: set,
    daily_done_set: set,
) -> float:
    """Calculate crypto funding fee for one symbol.

    Args:
        symbol: Instrument code.
        bar: Current bar data.
        timestamp: Bar timestamp.
        positions: Shared positions dict.
        funding_rate: Fixed rate per settlement.
        applied_set: (symbol, date, hour) dedup set — mutated.
        daily_done_set: (symbol, date) dedup set — mutated.

    Returns:
        Fee amount (positive = longs pay, negative = longs receive).
    """
    if not hasattr(timestamp, "date"):
        return 0.0

    current_date = timestamp.date()
    hour = timestamp.hour if hasattr(timestamp, "hour") else 0

    if hour in FUNDING_HOURS:
        key = (symbol, current_date, hour)
        if key in applied_set:
            return 0.0
        applied_set.add(key)
    else:
        day_key = (symbol, current_date)
        if day_key in daily_done_set:
            return 0.0
        daily_done_set.add(day_key)

    pos = positions.get(symbol)
    if pos is None:
        return 0.0

    mark_price = float(bar.get("close", pos.entry_price))
    notional = pos.size * mark_price
    return notional * funding_rate * pos.direction


def check_crypto_liquidation(
    symbol: str,
    bar: pd.Series,
    positions: Dict[str, Position],
) -> bool:
    """Check if a crypto position should be liquidated.

    Args:
        symbol: Instrument code.
        bar: Current bar data.
        positions: Shared positions dict.

    Returns:
        True if liquidation should be triggered.
        Does NOT execute the liquidation -- caller handles that.
    """
    pos = positions.get(symbol)
    if pos is None or pos.leverage <= 1.0:
        return False

    mark_price = float(bar.get("close", pos.entry_price))
    margin = pos.size * pos.entry_price / pos.leverage
    unrealized = pos.direction * pos.size * (mark_price - pos.entry_price)

    notional = pos.size * mark_price
    maint_rate = _maintenance_rate(notional)
    maint_margin = notional * maint_rate

    return (margin + unrealized) <= maint_margin


# ── Forex: swap tables ──

_SWAP_LONG: dict[str, float] = {
    "EUR/USD": -6.5, "GBP/USD": -3.0, "USD/JPY": 8.0, "USD/CHF": 4.0,
    "AUD/USD": -2.0, "USD/CAD": 2.0, "NZD/USD": -1.5,
}
_SWAP_SHORT: dict[str, float] = {
    "EUR/USD": 3.5, "GBP/USD": -1.0, "USD/JPY": -12.0, "USD/CHF": -8.0,
    "AUD/USD": -1.0, "USD/CAD": -5.0, "NZD/USD": -2.0,
}


def _normalize_symbol(symbol: str) -> str:
    """Normalize forex symbol to 'XXX/YYY' format."""
    s = symbol.replace(".FX", "").replace(".", "").strip()
    if "/" in s:
        return s.upper()
    if len(s) == 6:
        return f"{s[:3]}/{s[3:]}".upper()
    return s.upper()


def calc_forex_swap(
    symbol: str,
    timestamp: pd.Timestamp,
    positions: Dict[str, Position],
    lot_size: float,
    last_swap_dates: dict,
) -> float:
    """Calculate forex swap for one symbol.

    Args:
        symbol: Forex pair.
        timestamp: Bar timestamp.
        positions: Shared positions dict.
        lot_size: Standard lot size (e.g. 100_000).
        last_swap_dates: Per-symbol date tracking dict -- mutated.

    Returns:
        Swap amount (positive = credit, negative = debit).
    """
    if not hasattr(timestamp, "date"):
        return 0.0

    current_date = timestamp.date()
    if last_swap_dates.get(symbol) == current_date:
        return 0.0
    last_swap_dates[symbol] = current_date

    pos = positions.get(symbol)
    if pos is None:
        return 0.0

    pair = _normalize_symbol(symbol)
    lots = pos.size / lot_size

    if pos.direction == 1:
        swap_per_lot = _SWAP_LONG.get(pair, -1.0)
    else:
        swap_per_lot = _SWAP_SHORT.get(pair, -1.0)

    # Wednesday = triple swap (covers Sat+Sun)
    multiplier = 3.0 if timestamp.weekday() == 2 else 1.0
    return lots * swap_per_lot * multiplier

"""Base backtest engine with shared bar-by-bar execution loop.

All market engines inherit from BaseEngine and override market-rule methods.
The shared run_backtest() handles: data loading → signal generation →
pre-compute target weights (with optimizer) → bar-by-bar execution with
market rule enforcement → metrics → artifacts.
"""

from __future__ import annotations

import importlib
import json
import logging
import re as _re
import sys
from abc import ABC, abstractmethod
from collections.abc import Iterable
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

import pandas as pd

from backtest.loaders.rsshub_events import (
    FeedSpec,
    RSSHubEventProvider,
    enrich_price_frames_with_events,
    feed_specs_from_config,
)
from backtest.loaders.tushare_fundamentals import (
    TushareFundamentalProvider,
    enrich_price_frames_with_fundamentals,
)
from backtest.metrics import (
    by_exit_reason_stats,
    by_symbol_stats,
    calc_metrics,
)
from backtest.models import EquitySnapshot, Position, TradeRecord

logger = logging.getLogger(__name__)


def _run_card_data_sources(config: Dict[str, Any], loader: Any) -> List[str]:
    """Return source names for run-card evidence."""
    configured = config.get("_run_card_effective_sources")
    if isinstance(configured, list):
        return [str(source) for source in configured if str(source).strip()]
    if isinstance(configured, str) and configured.strip():
        return [configured.strip()]

    loader_name = getattr(loader, "name", None)
    if loader_name:
        return [str(loader_name)]

    source = config.get("source")
    return [str(source)] if source else []


# ─── Market detection (lightweight, for signal alignment only) ───

_CRYPTO_RE = _re.compile(r"^[A-Z]+-USDT$|^[A-Z]+/USDT$", _re.I)
_FOREX_RE = _re.compile(r"^[A-Z]{3}/[A-Z]{3}$|^[A-Z]{6}\.FX$")


def _detect_market_for_align(code: str) -> str:
    """Lightweight market detection for ffill_limit calculation."""
    if _CRYPTO_RE.match(code):
        return "crypto"
    if _FOREX_RE.match(code):
        return "forex"
    return "equity"


# ─── Signal alignment (reused from daily_portfolio logic) ───


def _align(
    data_map: Dict[str, pd.DataFrame],
    signal_map: Dict[str, pd.Series],
    codes: List[str],
    optimizer: Optional[Callable] = None,
) -> tuple:
    """Build aligned date index, close matrix, target-position matrix, return matrix.

    Signal is shifted by 1 bar (next-bar-open semantics) then normalised so
    ``sum(abs(weights)) <= 1.0``.

    Args:
        data_map: code -> OHLCV DataFrame.
        signal_map: code -> signal Series.
        codes: Valid instrument codes.
        optimizer: Optional weight optimiser ``(ret, pos, dates) -> pos``.

    Returns:
        (dates, close_df, positions_df, returns_df)
    """
    all_dates: set = set()
    for c in codes:
        all_dates.update(data_map[c].index)
    dates = pd.DatetimeIndex(sorted(all_dates))

    close = pd.DataFrame(index=dates, columns=codes, dtype=float)
    for c in codes:
        close[c] = data_map[c]["close"].reindex(dates)

    # ffill with limit to avoid masking long suspensions (e.g. 3-week halt)
    # Cross-market needs larger limit (Chinese New Year can be 9-10 bars)
    ffill_limit = 10 if len({_detect_market_for_align(c) for c in codes}) > 1 else 5
    close = close.ffill(limit=ffill_limit)

    # Drop symbols that are entirely NaN (no data overlap with date range)
    all_nan_cols = [c for c in codes if close[c].isna().all()]
    if all_nan_cols:
        logger.warning("Symbols dropped (no usable price data): %s", all_nan_cols)
        codes = [c for c in codes if c not in all_nan_cols]
        if not codes:
            raise ValueError("All symbols have no data in the requested date range")
        close = close[codes]

    pos = pd.DataFrame(0.0, index=dates, columns=codes)
    for c in codes:
        # Shift on each symbol's OWN trading calendar, then ffill to unified
        own_dates = data_map[c].index
        raw = signal_map[c].reindex(own_dates).fillna(0.0).clip(-1.0, 1.0)
        shifted = raw.shift(1).fillna(0.0)
        pos[c] = shifted.reindex(dates).ffill(limit=ffill_limit).fillna(0.0)

    ret = close.pct_change().fillna(0.0)

    if optimizer is not None:
        pos = optimizer(ret, pos, dates)

    scale = pos.abs().sum(axis=1).clip(lower=1.0)
    pos = pos.div(scale, axis=0)

    return dates, close, pos, ret


def _load_optimizer(config: Dict[str, Any]) -> Optional[Callable]:
    """Dynamically load an optimizer function from config.

    Args:
        config: Backtest configuration.

    Returns:
        Optimizer callable, or None.
    """
    opt_name = config.get("optimizer")
    if not opt_name:
        return None
    opt_params = config.get("optimizer_params") or {}
    try:
        mod = importlib.import_module(f"backtest.optimizers.{opt_name}")
        return lambda ret, pos, dates: mod.optimize(ret, pos, dates, **opt_params)
    except (ImportError, AttributeError) as e:
        print(f"[WARN] Failed to load optimizer '{opt_name}': {e}, falling back to equal weight")
        return None


def _normalise_fundamental_fields(config: Dict[str, Any]) -> dict[str, list[str]]:
    """Read the optional statement-table field map from backtest config."""
    raw_fields = config.get("fundamental_fields")
    if raw_fields in (None, {}):
        return {}
    if not isinstance(raw_fields, dict):
        raise ValueError("fundamental_fields must map table names to field-name lists")

    normalized: dict[str, list[str]] = {}
    for table, fields in raw_fields.items():
        if not isinstance(table, str) or not table.strip():
            raise ValueError("fundamental_fields table names must be non-empty strings")
        if fields is None:
            continue
        if isinstance(fields, str) or not isinstance(fields, Iterable):
            raise ValueError(f"fundamental_fields[{table!r}] must be a list of field names")

        field_list = list(fields)
        if not field_list:
            continue
        invalid = [field for field in field_list if not isinstance(field, str) or not field.strip()]
        if invalid:
            raise ValueError(f"fundamental_fields[{table!r}] contains invalid field names")
        normalized[table.strip()] = field_list
    return normalized


def _maybe_enrich_fundamentals(
    data_map: Dict[str, pd.DataFrame],
    config: Dict[str, Any],
) -> Dict[str, pd.DataFrame]:
    """Attach configured Tushare statement fields before signal generation."""
    fields_by_table = _normalise_fundamental_fields(config)
    if not fields_by_table:
        return data_map

    try:
        provider = TushareFundamentalProvider()
        return enrich_price_frames_with_fundamentals(
            data_map,
            provider,
            fields_by_table,
            as_of=config.get("end_date", ""),
            periods=config.get("fundamental_periods"),
        )
    except Exception as exc:
        raise RuntimeError(
            f"fundamental_fields requested but Tushare enrichment failed: {exc}"
        ) from exc


def _event_feed_specs(config: Dict[str, Any]) -> List[FeedSpec]:
    """Parse the optional ``event_feeds`` feed definitions from backtest config.

    ``event_feeds`` is a list of feed-definition dicts (there is no built-in
    catalogue) — each with ``name``/``route_template``/``event_type`` and an
    optional ``code_style``. An empty/absent value means "no event enrichment".
    """
    raw_feeds = config.get("event_feeds")
    if raw_feeds in (None, [], {}):
        return []
    if not isinstance(raw_feeds, (list, tuple)):
        raise ValueError("event_feeds must be a list of feed definitions")
    return feed_specs_from_config(raw_feeds)


def _maybe_enrich_events(
    data_map: Dict[str, pd.DataFrame],
    config: Dict[str, Any],
) -> Dict[str, pd.DataFrame]:
    """Attach a point-in-time-safe ``event_score`` column before signal generation."""
    specs = _event_feed_specs(config)
    if not specs:
        return data_map

    try:
        provider = RSSHubEventProvider(feeds=specs)
        if not provider.is_available():
            raise RuntimeError(f"RSSHub base URL not configured (set ${'RSSHUB_BASE_URL'})")
        return enrich_price_frames_with_events(
            data_map,
            provider,
            as_of=config.get("end_date", ""),
            decay_lambda=float(config.get("event_decay_lambda", 0.1)),
            lookback=int(config.get("event_lookback", 30)),
        )
    except Exception as exc:
        raise RuntimeError(
            f"event_feeds requested but RSSHub enrichment failed: {exc}"
        ) from exc


# ─── Base Engine ───


class BaseEngine(ABC):
    """Abstract base for all market engines.

    Subclasses override market-rule methods:
      - can_execute: whether a trade is allowed by market rules
      - round_size: lot-size rounding
      - calc_commission: fee structure
      - apply_slippage: slippage model
      - on_bar: per-bar hooks (funding fees, liquidation, etc.)
    """

    def __init__(self, config: dict):
        self.config = config
        self.initial_capital: float = config.get("initial_cash", 1_000_000)
        self.default_leverage: float = config.get("leverage", 1.0)
        self.capital: float = self.initial_capital
        self.positions: Dict[str, Position] = {}
        self.trades: List[TradeRecord] = []
        self.equity_snapshots: List[EquitySnapshot] = []
        self._bar_idx: int = 0
        self._active_symbol: str = ""  # set by _rebalance/_close_position for subclass use

    # ── Market rule interface (subclass must implement) ──

    @abstractmethod
    def can_execute(self, symbol: str, direction: int, bar: pd.Series) -> bool:
        """Whether market rules allow this trade.

        Args:
            symbol: Instrument identifier.
            direction: 1 (long), -1 (short), 0 (close).
            bar: Current bar data (OHLCV + extras).

        Returns:
            True if allowed.
        """

    @abstractmethod
    def round_size(self, raw_size: float, price: float) -> float:
        """Round position size per market lot rules.

        Args:
            raw_size: Desired size.
            price: Current price.

        Returns:
            Rounded size.
        """

    @abstractmethod
    def calc_commission(self, size: float, price: float, direction: int, is_open: bool) -> float:
        """Calculate commission for a trade.

        Args:
            size: Trade size.
            price: Execution price.
            direction: 1 or -1.
            is_open: True for opening, False for closing.

        Returns:
            Commission amount.
        """

    @abstractmethod
    def apply_slippage(self, price: float, direction: int) -> float:
        """Apply slippage to execution price.

        Args:
            price: Raw price.
            direction: 1 (buying / covering short) or -1 (selling / shorting).

        Returns:
            Slipped price.
        """

    def on_bar(self, symbol: str, bar: pd.Series, timestamp: pd.Timestamp) -> None:
        """Per-bar market-rule hook (funding fees, liquidation, etc.).

        Default: no-op. Override in subclass as needed.
        """

    # ── PnL / margin calculation hooks ──
    # Override in FuturesBaseEngine to inject contract multiplier.

    def _calc_pnl(
        self, symbol: str, direction: int, size: float,
        entry_price: float, exit_price: float,
    ) -> float:
        """Realised PnL for a closed position."""
        return direction * size * (exit_price - entry_price)

    def _calc_margin(
        self, symbol: str, size: float, price: float, leverage: float,
    ) -> float:
        """Margin (collateral) required for a position."""
        return size * price / leverage

    def _calc_raw_size(
        self, symbol: str, target_notional: float, price: float,
    ) -> float:
        """Convert target notional exposure to number of units/contracts."""
        return target_notional / price

    # ── Main entry ──

    def run_backtest(
        self,
        config: Dict[str, Any],
        loader: Any,
        signal_engine: Any,
        run_dir: Path,
        bars_per_year: int = 252,
    ) -> Dict[str, Any]:
        """Full backtest pipeline.

        Signature matches ``daily_portfolio.run_backtest`` for drop-in replacement.

        Args:
            config: Backtest configuration dict.
            loader: DataLoader with ``fetch()`` method.
            signal_engine: SignalEngine with ``generate()`` method.
            run_dir: Artifacts output directory.
            bars_per_year: Annualisation factor.

        Returns:
            Metrics dictionary.
        """
        codes = config.get("codes", [])
        interval = config.get("interval", "1D")
        extra_fields = config.get("extra_fields") or None

        # 1. Load data
        data_map = loader.fetch(
            codes,
            config.get("start_date", ""),
            config.get("end_date", ""),
            fields=extra_fields,
            interval=interval,
        )
        if not data_map:
            print(json.dumps({"error": "No data fetched"}))
            sys.exit(1)
        data_map = _maybe_enrich_fundamentals(data_map, config)
        data_map = _maybe_enrich_events(data_map, config)

        # 2. Generate signals
        signal_map = signal_engine.generate(data_map)
        if not isinstance(signal_map, dict):
            print(json.dumps({"error": (
                f"SignalEngine.generate() must return Dict[str, pd.Series], "
                f"got {type(signal_map).__name__}. "
                "Return a dict mapping symbol codes to pandas Series of signals."
            )}))
            sys.exit(1)
        for _code, _sig in signal_map.items():
            if not isinstance(_sig, pd.Series):
                print(json.dumps({"error": (
                    f"SignalEngine.generate() returned {type(_sig).__name__} for '{_code}', "
                    "expected pd.Series. Each value must be a pandas Series with DatetimeIndex."
                )}))
                sys.exit(1)
        valid_codes = sorted(c for c in signal_map if c in data_map)
        if not valid_codes:
            print(json.dumps({"error": "No valid signals generated"}))
            sys.exit(1)

        # 3. Pre-compute target weights (with optimizer)
        opt_fn = _load_optimizer(config)
        dates, close_df, target_pos, ret_df = _align(
            data_map, signal_map, valid_codes, optimizer=opt_fn,
        )

        # Sync codes after _align may have dropped all-NaN symbols
        valid_codes = [c for c in valid_codes if c in target_pos.columns]

        # 4. Bar-by-bar execution
        self._execute_bars(dates, data_map, close_df, target_pos, valid_codes)

        # 5. Build output series
        equity_series = pd.Series(
            [s.equity for s in self.equity_snapshots],
            index=[s.timestamp for s in self.equity_snapshots],
        )
        bench_ret = ret_df.mean(axis=1) if ret_df.shape[1] > 0 else pd.Series(0.0, index=dates)
        benchmark_metadata = {}

        # ── External benchmark fetch ──────────────────────────────────────────
        bench_ticker = config.get("benchmark")
        if bench_ticker and bench_ticker != "auto":
            from backtest.benchmark import resolve_benchmark
            bench_result = resolve_benchmark(
                strategy_codes=codes,
                source=config.get("source", "yfinance"),
                start_date=config.get("start_date", ""),
                end_date=config.get("end_date", ""),
                interval=interval,
                explicit=bench_ticker,
            )
            if bench_result is not None:
                bench_ret = bench_result.ret_series.reindex(dates).fillna(0.0)
                benchmark_metadata = {
                    "benchmark_ticker": bench_result.ticker,
                    "benchmark_return": bench_result.total_ret,
                }
        # ── External benchmark fetch ──────────────────────────────────────────

        bench_equity = self.initial_capital * (1 + bench_ret).cumprod()

        # 6. Metrics
        m = calc_metrics(equity_series, self.trades, self.initial_capital, bars_per_year, bench_ret)
        m.update(benchmark_metadata)
        m["by_symbol"] = by_symbol_stats(self.trades)
        m["by_exit_reason"] = by_exit_reason_stats(self.trades)

        # 7. Validation (optional — triggered by config["validation"])
        if config.get("validation"):
            from backtest.validation import run_validation
            v_results = run_validation(
                config, equity_series, self.trades, self.initial_capital, bars_per_year,
            )
            m["validation"] = v_results
            # Write validation.json artifact
            v_path = run_dir / "artifacts" / "validation.json"
            v_path.write_text(json.dumps(v_results, indent=2, ensure_ascii=False), encoding="utf-8")

        # 8. Artifacts
        self._write_artifacts(
            run_dir, data_map, dates, equity_series, bench_equity, bench_ret,
            target_pos, m, valid_codes,
        )

        # 9. Trust Layer run card
        from backtest.run_card import write_run_card
        write_run_card(
            run_dir,
            config,
            m,
            data_sources=_run_card_data_sources(config, loader),
            strategy_path=run_dir / "code" / "signal_engine.py",
        )

        # Print scalar metrics (skip nested dicts for JSON compat)
        print(json.dumps({k: v for k, v in m.items() if not isinstance(v, dict)}, indent=2))
        return m

    # ── Execution loop ──

    def _execute_bars(
        self,
        dates: pd.DatetimeIndex,
        data_map: Dict[str, pd.DataFrame],
        close_df: pd.DataFrame,
        target_pos: pd.DataFrame,
        codes: List[str],
    ) -> None:
        """Bar-by-bar execution with market rule enforcement."""
        for i, ts in enumerate(dates):
            self._bar_idx = i

            # a. Per-bar hooks (funding fees, liquidation checks)
            for c in codes:
                if ts in data_map[c].index:
                    self.on_bar(c, data_map[c].loc[ts], ts)

            # b. Rebalance each symbol to target weight
            equity = self._calc_equity(close_df, ts)
            for c in codes:
                try:
                    target_w = float(target_pos.at[ts, c]) if ts in target_pos.index else 0.0
                    self._rebalance(c, target_w, data_map.get(c), ts, equity)
                except Exception as exc:
                    logger.warning("Rebalance failed for %s at %s: %s", c, ts, exc)

            # c. Record equity snapshot
            snap_equity = self._calc_equity(close_df, ts)
            total_unrealized = 0.0
            for p in self.positions.values():
                cp = self._safe_price(close_df, ts, p.symbol, p.entry_price)
                total_unrealized += self._calc_pnl(p.symbol, p.direction, p.size, p.entry_price, cp)
            self.equity_snapshots.append(EquitySnapshot(
                timestamp=ts,
                capital=self.capital,
                unrealized=total_unrealized,
                equity=snap_equity,
                positions=len(self.positions),
            ))

        # d. Force close all remaining positions
        if len(dates) > 0:
            last_ts = dates[-1]
            for c in list(self.positions.keys()):
                price = self._safe_price(close_df, last_ts, c, self.positions[c].entry_price)
                self._close_position(c, price, last_ts, "end_of_backtest")

    def _calc_equity(self, close_df: pd.DataFrame, ts: pd.Timestamp) -> float:
        """Total equity = free cash + sum(margin + unrealised) per position."""
        equity = self.capital
        for sym, pos in self.positions.items():
            cp = self._safe_price(close_df, ts, sym, pos.entry_price)
            margin = self._calc_margin(sym, pos.size, pos.entry_price, pos.leverage)
            unrealized = self._calc_pnl(sym, pos.direction, pos.size, pos.entry_price, cp)
            equity += margin + unrealized
        return equity

    def _rebalance(
        self,
        symbol: str,
        target_weight: float,
        df: Optional[pd.DataFrame],
        ts: pd.Timestamp,
        equity: float,
    ) -> None:
        """Adjust position for *symbol* toward *target_weight*."""
        self._active_symbol = symbol
        target_dir = 1 if target_weight > 1e-9 else (-1 if target_weight < -1e-9 else 0)
        current_pos = self.positions.get(symbol)

        # Nothing to do
        if current_pos is None and target_dir == 0:
            return
        if df is None or ts not in df.index:
            return

        bar = df.loc[ts]

        # Close if target is flat or direction changed
        if current_pos is not None:
            need_close = target_dir == 0 or target_dir != current_pos.direction
            if need_close:
                if self.can_execute(symbol, 0, bar):
                    open_price = float(bar.get("open", bar.get("close", 0)))
                    price = self.apply_slippage(open_price, -current_pos.direction)
                    self._close_position(symbol, price, ts, "signal")
                else:
                    return  # blocked (e.g. limit-down can't sell)

        # Open new if target non-zero and no remaining position
        if target_dir != 0 and symbol not in self.positions:
            if not self.can_execute(symbol, target_dir, bar):
                return  # blocked (e.g. A-share no-short)

            open_price = float(bar.get("open", bar.get("close", 0)))
            if open_price <= 0:
                return

            slipped = self.apply_slippage(open_price, target_dir)
            leverage = self.default_leverage
            target_notional = abs(target_weight) * equity * leverage
            raw_size = self._calc_raw_size(symbol, target_notional, slipped)
            size = self.round_size(raw_size, slipped)
            if size <= 0:
                return

            margin = self._calc_margin(symbol, size, slipped, leverage)
            comm = self.calc_commission(size, slipped, target_dir, is_open=True)

            # Capital check — reduce if insufficient
            if margin + comm > self.capital:
                available = self.capital - comm
                if available <= 0:
                    return
                size = self.round_size(
                    self._calc_raw_size(symbol, available * leverage, slipped), slipped,
                )
                if size <= 0:
                    return
                margin = self._calc_margin(symbol, size, slipped, leverage)
                comm = self.calc_commission(size, slipped, target_dir, is_open=True)

            self.capital -= (margin + comm)
            self.positions[symbol] = Position(
                symbol=symbol,
                direction=target_dir,
                entry_price=slipped,
                entry_time=ts,
                size=size,
                leverage=leverage,
                entry_bar_idx=self._bar_idx,
                entry_commission=comm,
            )

    def _close_position(
        self,
        symbol: str,
        exit_price: float,
        exit_time: pd.Timestamp,
        reason: str,
    ) -> None:
        """Close position, record trade, return capital."""
        self._active_symbol = symbol
        pos = self.positions.pop(symbol, None)
        if pos is None:
            return

        pnl = self._calc_pnl(symbol, pos.direction, pos.size, pos.entry_price, exit_price)
        margin = self._calc_margin(symbol, pos.size, pos.entry_price, pos.leverage)
        pnl_pct = pnl / margin * 100 if margin > 1e-9 else 0.0
        exit_comm = self.calc_commission(pos.size, exit_price, pos.direction, is_open=False)

        self.capital += margin + pnl - exit_comm

        holding_bars = max(self._bar_idx - pos.entry_bar_idx, 0)

        self.trades.append(TradeRecord(
            symbol=symbol,
            direction=pos.direction,
            entry_price=pos.entry_price,
            exit_price=exit_price,
            entry_time=pos.entry_time,
            exit_time=exit_time,
            size=pos.size,
            leverage=pos.leverage,
            pnl=pnl,
            pnl_pct=pnl_pct,
            exit_reason=reason,
            holding_bars=holding_bars,
            commission=pos.entry_commission + exit_comm,
        ))

    # ── Artifacts ──

    def _write_artifacts(
        self,
        run_dir: Path,
        data_map: Dict[str, pd.DataFrame],
        dates: pd.DatetimeIndex,
        equity_series: pd.Series,
        bench_equity: pd.Series,
        bench_ret: pd.Series,
        target_pos: pd.DataFrame,
        metrics: dict,
        codes: List[str],
    ) -> None:
        """Write CSV artifacts compatible with daily_portfolio format."""
        out = run_dir / "artifacts"
        out.mkdir(parents=True, exist_ok=True)

        # OHLCV per symbol
        for code, df in data_map.items():
            df.to_csv(out / f"ohlcv_{code}.csv")

        # Equity curve
        port_ret = equity_series.pct_change().fillna(0.0)
        peak = equity_series.cummax()
        dd = (equity_series - peak) / peak.replace(0, 1)
        eq_df = pd.DataFrame({
            "ret": port_ret,
            "equity": equity_series,
            "drawdown": dd,
            "benchmark_equity": bench_equity.reindex(dates),
            "active_ret": port_ret - bench_ret.reindex(dates).fillna(0.0),
        }, index=dates)
        eq_df.index.name = "timestamp"
        eq_df.to_csv(out / "equity.csv")

        # Position weights (target, for compatibility)
        target_pos.index.name = "timestamp"
        target_pos.to_csv(out / "positions.csv")

        # Trades (compatible format)
        trade_rows = []
        for t in self.trades:
            # Entry event
            trade_rows.append({
                "timestamp": str(t.entry_time.date()) if hasattr(t.entry_time, "date") else str(t.entry_time),
                "code": t.symbol,
                "side": "buy" if t.direction == 1 else "sell",
                "price": round(t.entry_price, 4),
                "qty": round(t.size, 6),
                "reason": "signal",
                "pnl": 0.0,
                "holding_days": 0,
                "return_pct": 0.0,
            })
            # Exit event
            try:
                hold_days = (t.exit_time - t.entry_time).days
            except Exception:
                hold_days = 0
            trade_rows.append({
                "timestamp": str(t.exit_time.date()) if hasattr(t.exit_time, "date") else str(t.exit_time),
                "code": t.symbol,
                "side": "sell" if t.direction == 1 else "buy",
                "price": round(t.exit_price, 4),
                "qty": round(t.size, 6),
                "reason": t.exit_reason,
                "pnl": round(t.pnl, 4),
                "holding_days": hold_days,
                "return_pct": round(t.pnl_pct, 2),
            })

        trade_cols = ["timestamp", "code", "side", "price", "qty", "reason", "pnl", "holding_days", "return_pct"]
        pd.DataFrame(trade_rows or [], columns=trade_cols).to_csv(out / "trades.csv", index=False)

        # Metrics
        flat_metrics = {k: v for k, v in metrics.items() if not isinstance(v, dict)}
        pd.DataFrame([flat_metrics]).to_csv(out / "metrics.csv", index=False)

    # ── Helpers ──

    @staticmethod
    def _safe_price(
        close_df: pd.DataFrame,
        ts: pd.Timestamp,
        symbol: str,
        fallback: float,
    ) -> float:
        """Get close price with fallback."""
        if ts in close_df.index and symbol in close_df.columns:
            val = close_df.at[ts, symbol]
            if pd.notna(val):
                return float(val)
        return fallback

"""A-share (China mainland) backtest engine.

Market rules:
  - T+1: cannot sell shares bought today
  - No short selling for retail investors
  - Price limits: ±10% main board, ±20% ChiNext/STAR, ±5% ST
  - Minimum lot: 100 shares (odd lots can only be sold, not bought)
  - Commission: ¥5 minimum, 0.025% bilateral
  - Stamp tax: 0.05% sell-side only
  - Transfer fee: 0.001% bilateral
"""

from __future__ import annotations

import pandas as pd

from backtest.engines.base import BaseEngine


class ChinaAEngine(BaseEngine):
    """A-share market engine.

    Config keys:
      - commission_rate: default 0.00025 (万2.5)
      - commission_min: default 5.0 (RMB)
      - stamp_tax: default 0.0005 (万5, sell-only)
      - transfer_fee: default 0.00001 (万0.1)
      - slippage: default 0.001
    """

    def __init__(self, config: dict):
        config = {**config, "leverage": 1.0}  # A-shares: no leverage
        super().__init__(config)
        self.commission_rate: float = config.get("commission_rate", 0.00025)
        self.commission_min: float = config.get("commission_min", 5.0)
        self.stamp_tax: float = config.get("stamp_tax", 0.0005)
        self.transfer_fee: float = config.get("transfer_fee", 0.00001)
        self.slippage_rate: float = config.get("slippage", 0.001)

    def can_execute(self, symbol: str, direction: int, bar: pd.Series) -> bool:
        """A-share execution rules.

        Args:
            symbol: Stock code (e.g. 000001.SZ).
            direction: 1 (buy), -1 (short — always blocked), 0 (sell/close).
            bar: Current bar (needs 'close', 'pre_close' or 'pct_chg').

        Returns:
            True if the trade is allowed.
        """
        # 1. No short selling
        if direction == -1:
            return False

        # 2. T+1: can't sell shares bought today
        if direction == 0:
            pos = self.positions.get(symbol)
            if pos is not None:
                bar_date = _bar_date(bar)
                entry_date = pos.entry_time.date() if hasattr(pos.entry_time, "date") else None
                if bar_date is not None and entry_date is not None and bar_date == entry_date:
                    return False

        # 3. Price limits
        pct_chg = _calc_pct_change(bar)
        if pct_chg is not None:
            limit = _price_limit(symbol)
            if direction == 1 and pct_chg >= limit - 0.001:
                return False  # limit-up: can't buy
            if direction == 0 and pct_chg <= -limit + 0.001:
                return False  # limit-down: can't sell

        return True

    def round_size(self, raw_size: float, price: float) -> float:
        """Round down to 100-share lots."""
        return max(int(raw_size / 100) * 100, 0)

    def calc_commission(self, size: float, price: float, _direction: int, is_open: bool) -> float:
        """A-share fee structure: commission + stamp tax (sell) + transfer fee.

        ``_direction`` is unused today — reserved for future asymmetric
        long/short fee schedules (margin trading, securities lending).
        """
        notional = size * price
        # Commission: 万2.5, min ¥5
        comm = max(notional * self.commission_rate, self.commission_min)
        # Transfer fee: 万0.1 bilateral
        comm += notional * self.transfer_fee
        # Stamp tax: 万5 sell-only
        if not is_open:
            comm += notional * self.stamp_tax
        return comm

    def apply_slippage(self, price: float, direction: int) -> float:
        """A-share slippage (relatively small due to tick size)."""
        return price * (1 + direction * self.slippage_rate)


# ── Helpers ──


def _bar_date(bar: pd.Series):
    """Extract date from bar, handling various column names."""
    for col in ("trade_date", "date"):
        if col in bar.index:
            val = bar[col]
            if hasattr(val, "date"):
                return val.date()
            try:
                return pd.Timestamp(val).date()
            except Exception:
                pass
    # Fall back to bar name (index timestamp)
    if hasattr(bar, "name") and hasattr(bar.name, "date"):
        return bar.name.date()
    return None


# Note: china_futures and global_futures have variants that prioritise
# settle/pre_settle (futures-native); see those modules for the
# futures-specific logic.
def _calc_pct_change(bar: pd.Series):
    """Calculate price change percentage from bar data."""
    if "pct_chg" in bar.index:
        val = bar["pct_chg"]
        if pd.notna(val):
            return float(val) / 100.0  # tushare pct_chg is in percentage points

    close = bar.get("close")
    pre_close = bar.get("pre_close")
    if close is not None and pre_close is not None and pre_close > 0:
        return (float(close) - float(pre_close)) / float(pre_close)
    return None


def _price_limit(symbol: str) -> float:
    """Determine price limit based on board.

    Args:
        symbol: Stock code (e.g. 300001.SZ, 688001.SH, 000001.SZ).

    Returns:
        Limit as fraction (0.10, 0.20, or 0.05).
    """
    code = symbol.split(".")[0] if "." in symbol else symbol
    # ChiNext (300xxx) / STAR (688xxx): ±20%
    if code.startswith("300") or code.startswith("688"):
        return 0.20
    # ST stocks: ±5% (heuristic: can't fully detect from code alone)
    # Beijing exchange (8xxxxx): ±30% — simplified to 0.30
    if code.startswith("8") and len(code) == 6:
        return 0.30
    # Main board: ±10%
    return 0.10

"""China futures backtest engine.

Market rules (exchange-level, CFFEX / SHFE / DCE / ZCE / INE / GFEX):
  - T+0: can open and close same day (intraday trading allowed)
  - Margin trading: 5%~15% by product (exchange-set minimum)
  - Price limits: stock-index +-10%, bonds +-2%, commodities +-3%~8%
  - Commission: per-lot fixed or per-notional rate (varies by product)
  - Contract multiplier: product-specific (IF=300, rb=10, au=1000, ...)
  - Minimum trading unit: 1 contract
  - Night session: 21:00-02:30 (varies by product, not enforced in bar-level sim)
"""

from __future__ import annotations

import re

import pandas as pd

from backtest.engines.futures_base import FuturesBaseEngine


# ── Contract multiplier lookup ──

_MULTIPLIER: dict[str, int] = {
    # Stock index futures (CFFEX)
    "IF": 300, "IC": 200, "IH": 300, "IM": 200,
    # Treasury bond futures (CFFEX)
    "T": 10000, "TF": 10000, "TS": 20000, "TL": 10000,
    # Metals (SHFE)
    "au": 1000, "ag": 15, "cu": 5, "al": 5, "zn": 5,
    "pb": 5, "ni": 1, "sn": 1, "ss": 5,
    # Ferrous (SHFE / DCE)
    "rb": 10, "hc": 10, "i": 100, "j": 100, "jm": 60,
    # Energy (SHFE / INE)
    "sc": 1000, "fu": 10, "lu": 10, "bu": 10, "nr": 10,
    # Agriculture (DCE)
    "c": 10, "cs": 10, "m": 10, "y": 10, "a": 10,
    "p": 10, "jd": 10, "lh": 16, "rr": 10, "pg": 20,
    # Agriculture (ZCE)
    "CF": 5, "SR": 10, "TA": 5, "MA": 10, "AP": 10,
    "RM": 10, "OI": 10, "CJ": 5, "PK": 5, "CY": 5,
    # Chemical (DCE / ZCE)
    "pp": 5, "l": 5, "v": 5, "eg": 10, "eb": 5,
    "PF": 5, "SA": 20, "FG": 20, "UR": 20,
    # GFEX
    "si": 5, "lc": 1,
}

# ── Margin rate (exchange minimum) ──

_MARGIN_RATE: dict[str, float] = {
    # CFFEX stock index
    "IF": 0.12, "IC": 0.12, "IH": 0.12, "IM": 0.12,
    # CFFEX bonds
    "T": 0.03, "TF": 0.02, "TS": 0.015, "TL": 0.035,
    # SHFE metals
    "au": 0.08, "ag": 0.09, "cu": 0.08, "al": 0.07,
    "zn": 0.08, "pb": 0.08, "ni": 0.12, "sn": 0.10, "ss": 0.08,
    # Ferrous
    "rb": 0.10, "hc": 0.10, "i": 0.12, "j": 0.12, "jm": 0.12,
    # Energy
    "sc": 0.10, "fu": 0.10, "lu": 0.10, "bu": 0.10,
    # Agriculture
    "c": 0.07, "cs": 0.07, "m": 0.08, "y": 0.08, "a": 0.08,
    "p": 0.08, "jd": 0.08, "lh": 0.12,
    # Textiles / chemical
    "CF": 0.07, "SR": 0.07, "TA": 0.07, "MA": 0.07,
    "pp": 0.07, "l": 0.07, "v": 0.07, "eg": 0.08,
    "SA": 0.08, "FG": 0.08, "UR": 0.08,
}

# ── Price limit (fraction, ± from settlement) ──

_PRICE_LIMIT: dict[str, float] = {
    # CFFEX stock index ±10%
    "IF": 0.10, "IC": 0.10, "IH": 0.10, "IM": 0.10,
    # CFFEX bonds ±2% (simplified)
    "T": 0.02, "TF": 0.012, "TS": 0.005, "TL": 0.035,
}
_DEFAULT_PRICE_LIMIT = 0.05  # most commodities ±4%~7%, use 5% as default

# ── Commission structure ──
# ("rate", pct) = per-notional  |  ("fixed", amount_per_lot) = per-contract

_COMMISSION: dict[str, tuple[str, float]] = {
    # CFFEX stock index: ~0.0023% of notional
    "IF": ("rate", 0.000023), "IC": ("rate", 0.000023),
    "IH": ("rate", 0.000023), "IM": ("rate", 0.000023),
    # CFFEX bonds
    "T": ("fixed", 3.0), "TF": ("fixed", 3.0), "TS": ("fixed", 3.0),
    # Metals
    "au": ("fixed", 10.0), "ag": ("fixed", 3.0), "cu": ("fixed", 5.0),
    "al": ("fixed", 3.0), "zn": ("fixed", 3.0), "ni": ("fixed", 3.0),
    # Ferrous
    "rb": ("rate", 0.0001), "hc": ("rate", 0.0001), "i": ("rate", 0.0001),
    "j": ("rate", 0.0001), "jm": ("rate", 0.0001),
    # Energy
    "sc": ("fixed", 20.0), "fu": ("rate", 0.00005),
    # Agriculture
    "c": ("fixed", 1.2), "cs": ("fixed", 1.5), "m": ("fixed", 1.5),
    "y": ("fixed", 2.5), "a": ("fixed", 2.0), "p": ("fixed", 2.5),
    "jd": ("rate", 0.00015), "lh": ("rate", 0.0002),
    # Textiles / chemical
    "CF": ("fixed", 4.3), "SR": ("fixed", 3.0), "TA": ("fixed", 3.0),
    "MA": ("fixed", 2.0), "pp": ("fixed", 1.0), "l": ("fixed", 1.0),
    "v": ("fixed", 1.0), "SA": ("fixed", 3.5), "FG": ("fixed", 3.0),
}
_DEFAULT_COMMISSION: tuple[str, float] = ("fixed", 5.0)


def _extract_product(symbol: str) -> str:
    """Extract product code from futures symbol.

    Examples:
        'IF2406.CFFEX' -> 'IF'
        'rb2410.SHFE'  -> 'rb'
        'au2412'       -> 'au'

    Args:
        symbol: Futures symbol string.

    Returns:
        Product code (e.g. 'IF', 'rb', 'au').
    """
    code = symbol.split(".")[0]
    m = re.match(r"([A-Za-z]+)", code)
    return m.group(1) if m else code


class ChinaFuturesEngine(FuturesBaseEngine):
    """China futures engine covering CFFEX / SHFE / DCE / ZCE / INE / GFEX.

    Config keys:
      - slippage: default 0.0005
      - margin_rate_override: override margin rate for all products
      - commission_override: override commission for all products
    """

    def __init__(self, config: dict):
        # Derive leverage from margin rate of first code, or use config override
        margin_override = config.get("margin_rate_override")
        if margin_override:
            leverage = 1.0 / margin_override
        else:
            codes = config.get("codes", [])
            if codes:
                product = _extract_product(codes[0])
                mr = _MARGIN_RATE.get(product, 0.10)
                leverage = 1.0 / mr
            else:
                leverage = 10.0  # ~10% margin default
        config = {**config, "leverage": leverage}
        super().__init__(config)
        self.slippage_rate: float = config.get("slippage", 0.0005)
        self._commission_override = config.get("commission_override")

    def can_execute(self, symbol: str, direction: int, bar: pd.Series) -> bool:
        """China futures: T+0, both directions, price-limit enforced.

        Args:
            symbol: Futures code.
            direction: 1 (long), -1 (short), 0 (close).
            bar: Current bar data.

        Returns:
            True if allowed.
        """
        # T+0: no same-day sell restriction
        # Both long and short allowed

        # Price limit check
        pct_chg = _calc_pct_change(bar)
        if pct_chg is not None:
            product = _extract_product(symbol)
            limit = _PRICE_LIMIT.get(product, _DEFAULT_PRICE_LIMIT)
            if direction == 1 and pct_chg >= limit - 0.001:
                return False  # limit-up: can't open long / can't buy
            if direction == -1 and pct_chg <= -limit + 0.001:
                return False  # limit-down: can't open short
            if direction == 0:
                pos = self.positions.get(symbol)
                if pos is not None:
                    # Can't close long at limit-down, can't close short at limit-up
                    if pos.direction == 1 and pct_chg <= -limit + 0.001:
                        return False
                    if pos.direction == -1 and pct_chg >= limit - 0.001:
                        return False
        return True

    def round_size(self, raw_size: float, price: float) -> float:
        """Minimum 1 contract, integer lots only."""
        return max(int(raw_size), 0)

    def calc_commission(self, size: float, price: float, _direction: int, is_open: bool) -> float:
        """Commission varies by product: fixed per-lot or percentage of notional.

        ``_direction`` is unused — reserved for future open/close-fee
        asymmetry (some products charge different rates for close-today).
        """
        if self._commission_override is not None:
            return size * price * self._commission_override
        return self.calc_commission_for_symbol(self._active_symbol, size, price, is_open)

    def calc_commission_for_symbol(
        self, symbol: str, size: float, price: float, is_open: bool,
    ) -> float:
        """Symbol-aware commission calculation.

        Args:
            symbol: Futures code.
            size: Number of contracts.
            price: Execution price.
            is_open: True for opening trade.

        Returns:
            Commission in RMB.
        """
        product = _extract_product(symbol)
        mode, value = _COMMISSION.get(product, _DEFAULT_COMMISSION)
        cm = _MULTIPLIER.get(product, 10)
        if mode == "rate":
            return size * price * cm * value
        return size * value

    def apply_slippage(self, price: float, direction: int) -> float:
        """Futures slippage."""
        return price * (1 + direction * self.slippage_rate)

    def get_contract_multiplier(self, symbol: str) -> float:
        """Look up contract multiplier from product code."""
        product = _extract_product(symbol)
        return float(_MULTIPLIER.get(product, 10))

    def get_margin_rate(self, symbol: str) -> float:
        """Look up exchange margin rate for a product.

        Args:
            symbol: Futures symbol.

        Returns:
            Margin rate (e.g. 0.10 for 10%).
        """
        product = _extract_product(symbol)
        return _MARGIN_RATE.get(product, 0.10)


# ── Helpers ──


# Note: china_a uses close/pre_close-only; global_futures prioritises
# close/pre_close before settle. This China-futures variant prefers
# settle/pre_settle because tushare reports settlement as the canonical
# daily price for domestic contracts. See those modules for the equity /
# global-futures equivalents.
def _calc_pct_change(bar: pd.Series):
    """Calculate bar price change percentage.

    Priority: settle/pre_settle (futures native) > close/pre_close > pct_chg.
    pct_chg from tushare is always in percentage points (0.5 = 0.5%).
    """
    # Prefer settlement prices (unambiguous for futures)
    settle = bar.get("settle")
    pre_settle = bar.get("pre_settle")
    if settle is not None and pre_settle is not None and pre_settle > 0:
        return (float(settle) - float(pre_settle)) / float(pre_settle)

    close = bar.get("close")
    pre_close = bar.get("pre_close")
    if close is not None and pre_close is not None and pre_close > 0:
        return (float(close) - float(pre_close)) / float(pre_close)

    # tushare pct_chg is in percentage points (e.g. 0.5 = 0.5%)
    if "pct_chg" in bar.index:
        val = bar["pct_chg"]
        if pd.notna(val):
            return float(val) / 100.0

    return None