Bybitデータでバックテスト基盤を自作するPython実装
手数料・スリッページ・資金調達率・清算を織り込む

"""costs.py — 自分の口座の実際の手数料率と、銘柄の資金調達率を取得する
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

import pandas as pd
from pybit.unified_trading import HTTP


@dataclass(frozen=True)
class FeeModel:
    maker: Decimal      # 例: Decimal("0.0002") = 0.02%
    taker: Decimal      # 例: Decimal("0.00055") = 0.055%

    def cost(self, notional: Decimal, is_maker: bool) -> Decimal:
        """約定金額に対する手数料。マイナス（リベート）もありうる。"""
        rate = self.maker if is_maker else self.taker
        return notional * rate

    def round_trip_bps(self, maker_ratio: float = 0.0) -> float:
        """往復コストをbpsで返す。maker_ratio=1.0なら全部メイカー。"""
        rate = float(self.maker) * maker_ratio + float(self.taker) * (1 - maker_ratio)
        return rate * 2 * 10_000


def fetch_fee_model(http: HTTP, symbol: str = "BTCUSDT",
                    category: str = "linear") -> FeeModel:
    """自分の口座に適用される手数料率を取得する（要APIキー）。"""
    res = http.get_fee_rates(category=category, symbol=symbol)
    if res["retCode"] != 0:
        raise RuntimeError(f"get_fee_rates失敗: {res['retMsg']}")
    row = res["result"]["list"][0]
    return FeeModel(maker=Decimal(row["makerFeeRate"]),
                    taker=Decimal(row["takerFeeRate"]))


def fetch_funding_history(http: HTTP, symbol: str, start_ms: int,
                          category: str = "linear") -> pd.DataFrame:
    """資金調達率の履歴を全件取得する（1回200件・新しい順）。"""
    rows: list[dict] = []
    end = None
    for _ in range(200):                       # 安全上限
        kw = dict(category=category, symbol=symbol, limit=200)
        if end:
            kw["endTime"] = end
        res = http.get_funding_rate_history(**kw)
        if res["retCode"] != 0:
            raise RuntimeError(f"get_funding_rate_history失敗: {res['retMsg']}")
        page = res["result"]["list"]
        if not page:
            break
        rows.extend(page)
        oldest = int(page[-1]["fundingRateTimestamp"])
        if oldest <= start_ms:
            break
        end = oldest - 1

    if not rows:
        return pd.DataFrame(columns=["ts", "funding_rate"]).set_index("ts")

    df = pd.DataFrame(rows)
    df["ts"] = pd.to_datetime(df["fundingRateTimestamp"].astype("int64"),
                              unit="ms", utc=True)
    df["funding_rate"] = df["fundingRate"].astype(float)
    return (df[["ts", "funding_rate"]]
            .drop_duplicates(subset="ts")
            .set_index("ts").sort_index())


def funding_interval_hours(http: HTTP, symbol: str,
                           category: str = "linear") -> float:
    """収受間隔(時間)。銘柄によって8h/4h/1hなど異なるので必ず取得する。"""
    res = http.get_instruments_info(category=category, symbol=symbol)
    minutes = int(res["result"]["list"][0]["fundingInterval"])
    return minutes / 60


if __name__ == "__main__":
    from client import make_client

    c = make_client()
    fees = fetch_fee_model(c.http, "BTCUSDT")
    print(f"maker={float(fees.maker):.5%}  taker={float(fees.taker):.5%}")
    print(f"往復コスト(全テイカー): {fees.round_trip_bps(0.0):.2f} bps")
    print(f"往復コスト(全メイカー): {fees.round_trip_bps(1.0):.2f} bps")

    hours = funding_interval_hours(c.http, "BTCUSDT")
    fund = fetch_funding_history(c.http, "BTCUSDT",
                                 start_ms=int(pd.Timestamp("2025-01-01").timestamp() * 1000))
    ann = fund["funding_rate"].mean() * (24 / hours) * 365
    print(f"\n資金調達: 収受間隔{hours:.0f}h  平均{fund['funding_rate'].mean():+.6%}  "
          f"年率換算 {ann:+.2%}  ({len(fund)}件)")

"""bt/types.py — バックテストと本番で共有する型
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum

import pandas as pd


class Side(str, Enum):
    BUY = "Buy"
    SELL = "Sell"

    @property
    def sign(self) -> int:
        return 1 if self is Side.BUY else -1

    @property
    def opposite(self) -> "Side":
        return Side.SELL if self is Side.BUY else Side.BUY


class OrderType(str, Enum):
    MARKET = "Market"
    LIMIT = "Limit"
    POST_ONLY = "PostOnly"       # 指値かつメイカー確定（約定しないことがある）


@dataclass
class Bar:
    ts: pd.Timestamp
    open: float
    high: float
    low: float
    close: float
    volume: float
    turnover: float = 0.0
    mark_close: float | None = None      # 清算判定用（記事07で取得）


@dataclass
class Order:
    side: Side
    qty: float
    order_type: OrderType = OrderType.MARKET
    price: float | None = None
    reduce_only: bool = False
    tag: str = ""
    ttl_bars: int | None = None          # 何本で失効させるか（Noneで無期限）
    # --- エンジンが埋める ---
    placed_at: pd.Timestamp | None = None
    bars_alive: int = 0


@dataclass
class Fill:
    ts: pd.Timestamp
    side: Side
    qty: float
    price: float
    fee: float
    is_maker: bool
    tag: str = ""
    reason: str = "signal"               # signal / liquidation / ttl など


@dataclass
class Position:
    """一方向のみ保持するシンプルなポジション（ワンウェイ想定）。"""

    qty: float = 0.0                     # 正=ロング、負=ショート
    entry_price: float = 0.0

    @property
    def is_flat(self) -> bool:
        return abs(self.qty) < 1e-12

    @property
    def side(self) -> Side | None:
        if self.is_flat:
            return None
        return Side.BUY if self.qty > 0 else Side.SELL

    def notional(self, price: float) -> float:
        return abs(self.qty) * price

    def unrealised(self, price: float) -> float:
        return (price - self.entry_price) * self.qty

    def apply(self, side: Side, qty: float, price: float) -> float:
        """約定を適用し、実現損益を返す。"""
        signed = qty * side.sign
        realised = 0.0

        if self.is_flat or (self.qty > 0) == (signed > 0):
            # 新規 or 増し玉 -> 加重平均で建値を更新
            total = self.qty + signed
            if abs(total) > 1e-12:
                self.entry_price = (
                    (self.entry_price * self.qty + price * signed) / total
                )
            self.qty = total
        else:
            # 反対売買 -> 決済分の損益を確定
            closing = min(abs(signed), abs(self.qty))
            realised = (price - self.entry_price) * closing * (1 if self.qty > 0 else -1)
            self.qty += signed
            if self.is_flat:
                self.qty, self.entry_price = 0.0, 0.0
            elif (self.qty > 0) != (signed < 0):
                # ドテン（反対側に転換）した場合は建値を約定価格に
                self.entry_price = price
        return realised


@dataclass
class Snapshot:
    ts: pd.Timestamp
    equity: float
    cash: float
    position_qty: float
    price: float
    unrealised: float = 0.0
    funding_paid: float = 0.0
    fee_paid: float = 0.0
    events: list[str] = field(default_factory=list)

# 買い指値: その足の安値が指値以下なら約定した、とみなす
if bar.low <= limit_price:
    fill(limit_price)          # <- 楽観的すぎる

"""bt/fill.py — 約定モデル。楽観的な約定を構造的に避ける
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import random
from dataclasses import dataclass

from .types import Bar, Order, OrderType, Side


@dataclass
class FillModel:
    """約定判定と約定価格の決定。

    slippage_bps      : 成行の不利側スリッページ（bps）。実測値を入れる
    limit_needs_cross : True なら指値は「突き抜け」が必要（タッチでは約定しない）
    maker_fill_prob   : 突き抜けた場合の約定確率。記事13の実測値を入れる
    seed              : 再現性のための乱数シード
    """

    slippage_bps: float = 2.0
    limit_needs_cross: bool = True
    maker_fill_prob: float = 0.7
    seed: int = 42

    def __post_init__(self) -> None:
        self._rng = random.Random(self.seed)

    # ---- 成行 ----------------------------------------------------------
    def market_price(self, bar: Bar, side: Side) -> float:
        """成行は「次の足の始値 + 不利側スリッページ」で約定させる。

        ルックアヘッドを避けるため、判断した足の終値では約定させない。
        """
        slip = bar.open * self.slippage_bps / 10_000
        return bar.open + slip * side.sign      # 買いは高く、売りは安く

    # ---- 指値 ----------------------------------------------------------
    def limit_filled(self, bar: Bar, order: Order) -> tuple[bool, float]:
        assert order.price is not None
        px = order.price

        if order.side is Side.BUY:
            touched = bar.low <= px
            crossed = bar.low < px
        else:
            touched = bar.high >= px
            crossed = bar.high > px

        if self.limit_needs_cross:
            if not crossed:
                return False, 0.0
        elif not touched:
            return False, 0.0

        # 突き抜けても行列の後ろなら約定しない。確率でモデル化する
        if order.order_type is OrderType.POST_ONLY:
            if self._rng.random() > self.maker_fill_prob:
                return False, 0.0

        # ギャップで飛んだ場合は、より不利な始値で約定する
        if order.side is Side.BUY and bar.open < px:
            return True, bar.open
        if order.side is Side.SELL and bar.open > px:
            return True, bar.open
        return True, px

    def is_maker(self, order: Order) -> bool:
        return order.order_type in (OrderType.LIMIT, OrderType.POST_ONLY)


@dataclass
class PessimisticFillModel(FillModel):
    """感度分析用の悲観シナリオ。これでも生き残る戦略だけを採用する。"""

    slippage_bps: float = 8.0
    limit_needs_cross: bool = True
    maker_fill_prob: float = 0.4

"""bt/engine.py — イベントドリブンのバックテストエンジン
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

設計の要点:
  - Strategyには「確定した足まで」しか渡さない（ルックアヘッドの構造的防止）
  - 発注は次の足で処理される（同じ足での判断と約定を禁止）
"""
from __future__ import annotations

from dataclasses import dataclass, field

import pandas as pd

from costs import FeeModel
from .fill import FillModel
from .types import Bar, Fill, Order, OrderType, Position, Side, Snapshot


@dataclass
class BacktestConfig:
    initial_cash: float = 10_000.0
    leverage_cap: float = 5.0
    maintenance_margin_rate: float = 0.005   # 0.5%（銘柄・リスクリミットで変わる）
    funding_interval_hours: float = 8.0
    allow_liquidation: bool = True


@dataclass
class BacktestResult:
    equity: pd.Series
    fills: list[Fill]
    snapshots: list[Snapshot]
    liquidated: bool = False
    stats: dict = field(default_factory=dict)


class Backtester:
    def __init__(self, bars: pd.DataFrame, fees: FeeModel,
                 fill_model: FillModel | None = None,
                 funding: pd.Series | None = None,
                 config: BacktestConfig | None = None):
        self._validate(bars)
        self.bars = bars
        self.fees = fees
        self.fill = fill_model or FillModel()
        self.funding = funding            # index=UTC時刻, value=資金調達率
        self.cfg = config or BacktestConfig()

    # ---- 入力の健全性チェック（ここで落とすのが最も安い） ---------------
    @staticmethod
    def _validate(bars: pd.DataFrame) -> None:
        if not isinstance(bars.index, pd.DatetimeIndex):
            raise TypeError("indexがDatetimeIndexではありません")
        if bars.index.tz is None:
            raise ValueError("indexがtz-naiveです。UTCを明示してください")
        if not bars.index.is_monotonic_increasing:
            raise ValueError("indexが昇順ではありません（未来参照の危険）")
        if bars.index.duplicated().any():
            raise ValueError("重複した足があります")
        if not (bars["high"] >= bars["low"]).all():
            raise ValueError("high < low の足があります（データ破損）")
        for col in ("open", "high", "low", "close", "volume"):
            if col not in bars.columns:
                raise ValueError(f"列 {col} がありません")

    # ---- 本体 ----------------------------------------------------------
    def run(self, strategy) -> BacktestResult:
        cash = self.cfg.initial_cash
        pos = Position()
        pending: list[Order] = []
        fills: list[Fill] = []
        snapshots: list[Snapshot] = []
        equity_curve: list[tuple[pd.Timestamp, float]] = []

        total_fee = 0.0
        total_funding = 0.0
        liquidated = False

        rows = self.bars.itertuples()
        strategy.on_start(self.bars.iloc[:0])

        for i, row in enumerate(rows):
            bar = Bar(ts=row.Index, open=row.open, high=row.high, low=row.low,
                      close=row.close, volume=row.volume,
                      turnover=getattr(row, "turnover", 0.0),
                      mark_close=getattr(row, "mark_close", None))
            events: list[str] = []

            # ---- (1) 前の足で出された注文を、この足で処理する ----------
            still_pending: list[Order] = []
            for order in pending:
                if order.order_type is OrderType.MARKET:
                    price = self.fill.market_price(bar, order.side)
                    ok = True
                else:
                    ok, price = self.fill.limit_filled(bar, order)

                if not ok:
                    order.bars_alive += 1
                    if order.ttl_bars is not None and order.bars_alive >= order.ttl_bars:
                        events.append(f"expire:{order.tag}")
                    else:
                        still_pending.append(order)
                    continue

                qty = order.qty
                if order.reduce_only:
                    qty = min(qty, abs(pos.qty))
                    if qty <= 0:
                        continue

                is_maker = self.fill.is_maker(order)
                notional = qty * price
                fee = float(self.fees.cost(
                    __import__("decimal").Decimal(str(notional)), is_maker))

                realised = pos.apply(order.side, qty, price)
                cash += realised - fee
                total_fee += fee

                f = Fill(ts=bar.ts, side=order.side, qty=qty, price=price,
                         fee=fee, is_maker=is_maker, tag=order.tag)
                fills.append(f)
                strategy.on_fill(f)
                events.append(f"fill:{order.side.value}:{qty:.6f}@{price:.2f}")

            pending = still_pending

            # ---- (2) 資金調達の課金（該当時刻を跨いだら） --------------
            if self.funding is not None and not pos.is_flat:
                paid = self._apply_funding(bar, pos, i)
                if paid:
                    cash -= paid
                    total_funding += paid
                    events.append(f"funding:{-paid:+.4f}")

            # ---- (3) 清算判定（マーク価格ベース） ----------------------
            equity = cash + pos.unrealised(bar.close)
            if self.cfg.allow_liquidation and not pos.is_flat:
                if self._is_liquidated(bar, pos, cash):
                    price = bar.mark_close or bar.close
                    realised = pos.apply(pos.side.opposite, abs(pos.qty), price)
                    cash += realised
                    fills.append(Fill(ts=bar.ts, side=pos.side or Side.SELL,
                                      qty=abs(pos.qty), price=price, fee=0.0,
                                      is_maker=False, reason="liquidation"))
                    pos = Position()
                    liquidated = True
                    events.append("LIQUIDATED")
                    equity = cash
                    equity_curve.append((bar.ts, equity))
                    break                       # 清算で終了

            equity_curve.append((bar.ts, equity))
            snapshots.append(Snapshot(
                ts=bar.ts, equity=equity, cash=cash, position_qty=pos.qty,
                price=bar.close, unrealised=pos.unrealised(bar.close),
                funding_paid=total_funding, fee_paid=total_fee, events=events,
            ))

            # ---- (4) この足が確定したので、Strategyに判断させる --------
            #      ここで渡すのは「i番目まで」＝未来を含まない
            history = self.bars.iloc[: i + 1]
            new_orders = strategy.on_bar(history, pos, equity) or []
            for o in new_orders:
                o.placed_at = bar.ts
                pending.append(o)               # 次の足で処理される

        eq = pd.Series(dict(equity_curve), name="equity")
        eq.index = pd.DatetimeIndex(eq.index, tz="UTC")

        return BacktestResult(
            equity=eq, fills=fills, snapshots=snapshots, liquidated=liquidated,
            stats={"total_fee": total_fee, "total_funding": total_funding},
        )

    # ---- 補助 ----------------------------------------------------------
    def _apply_funding(self, bar: Bar, pos: Position, i: int) -> float:
        """この足の区間に資金調達の収受時刻が含まれていれば課金する。

        ロングは資金調達率が正のとき支払い、負のとき受取り。
        """
        if i == 0:
            return 0.0
        prev_ts = self.bars.index[i - 1]
        window = self.funding.loc[
            (self.funding.index > prev_ts) & (self.funding.index <= bar.ts)
        ]
        if window.empty:
            return 0.0
        rate = float(window.sum())
        notional = pos.notional(bar.close)
        # ロング(qty>0)は正のレートで支払い。ショートは受取り。
        return notional * rate * (1 if pos.qty > 0 else -1)

    def _is_liquidated(self, bar: Bar, pos: Position, cash: float) -> bool:
        """維持証拠金を割ったかを、その足の最悪値で判定する。"""
        worst = bar.low if pos.qty > 0 else bar.high
        if bar.mark_close is not None:
            # マーク価格が取れる場合は乖離を考慮して保守的に寄せる
            worst = min(worst, bar.mark_close) if pos.qty > 0 else max(worst, bar.mark_close)
        equity_worst = cash + (worst - pos.entry_price) * pos.qty
        maint = pos.notional(worst) * self.cfg.maintenance_margin_rate
        return equity_worst <= maint

"""bt/strategy.py — バックテストと本番で共有するStrategy基底クラス
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

from abc import ABC, abstractmethod

import pandas as pd

from .types import Fill, Order, OrderType, Position, Side


class Strategy(ABC):
    """on_barは「確定した足の履歴」だけを受け取る。

    バックテストではエンジンが履歴をスライスして渡し、
    本番ではWebSocketの確定足（記事26）で同じ形の履歴を渡す。
    このためStrategyの実装は環境を意識しなくてよい。
    """

    def on_start(self, history: pd.DataFrame) -> None:
        pass

    @abstractmethod
    def on_bar(self, history: pd.DataFrame, position: Position,
               equity: float) -> list[Order] | None:
        """確定足ごとに呼ばれる。返したOrderは「次の足」で処理される。"""

    def on_fill(self, fill: Fill) -> None:
        pass


class SmaCrossStrategy(Strategy):
    """SMAクロスの最小例。動作確認用であり、収益性を主張するものではない。"""

    def __init__(self, fast: int = 20, slow: int = 60, risk_pct: float = 0.5,
                 use_post_only: bool = False):
        if fast >= slow:
            raise ValueError("fast < slow である必要があります")
        self.fast, self.slow = fast, slow
        self.risk_pct = risk_pct
        self.use_post_only = use_post_only

    def on_bar(self, history: pd.DataFrame, position: Position,
               equity: float) -> list[Order] | None:
        if len(history) < self.slow + 1:
            return None                     # ウォームアップ中

        close = history["close"]
        fast_now = close.iloc[-self.fast:].mean()
        slow_now = close.iloc[-self.slow:].mean()
        fast_prev = close.iloc[-self.fast - 1:-1].mean()
        slow_prev = close.iloc[-self.slow - 1:-1].mean()

        crossed_up = fast_prev <= slow_prev and fast_now > slow_now
        crossed_down = fast_prev >= slow_prev and fast_now < slow_now

        price = close.iloc[-1]
        target_notional = equity * self.risk_pct
        qty = round(target_notional / price, 6)
        if qty <= 0:
            return None

        otype = OrderType.POST_ONLY if self.use_post_only else OrderType.MARKET
        limit_px = price if self.use_post_only else None
        orders: list[Order] = []

        if crossed_up and position.qty <= 0:
            if position.qty < 0:
                orders.append(Order(side=Side.BUY, qty=abs(position.qty),
                                    order_type=OrderType.MARKET,
                                    reduce_only=True, tag="exit-short"))
            orders.append(Order(side=Side.BUY, qty=qty, order_type=otype,
                                price=limit_px, ttl_bars=3, tag="entry-long"))
        elif crossed_down and position.qty >= 0:
            if position.qty > 0:
                orders.append(Order(side=Side.SELL, qty=position.qty,
                                    order_type=OrderType.MARKET,
                                    reduce_only=True, tag="exit-long"))
            orders.append(Order(side=Side.SELL, qty=qty, order_type=otype,
                                price=limit_px, ttl_bars=3, tag="entry-short"))
        return orders or None

"""bt/report.py — 成績指標の算出とレポート出力
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from .engine import BacktestResult

# 年間の足数。1時間足なら 24*365
BARS_PER_YEAR = {"1min": 525_600, "5min": 105_120, "15min": 35_040,
                 "1h": 8_760, "4h": 2_190, "1D": 365}


def max_drawdown(equity: pd.Series) -> tuple[float, pd.Timestamp | None, int]:
    """最大ドローダウン率、発生時刻、最長水没期間（本数）を返す。"""
    peak = equity.cummax()
    dd = equity / peak - 1.0
    trough = dd.idxmin() if len(dd) else None

    under = (equity < peak).astype(int)
    longest, run = 0, 0
    for v in under:
        run = run + 1 if v else 0
        longest = max(longest, run)
    return float(dd.min()) if len(dd) else 0.0, trough, longest


def metrics(result: BacktestResult, bars_per_year: int = 8_760) -> dict:
    eq = result.equity.dropna()
    if len(eq) < 3:
        return {"error": "データ不足"}

    ret = eq.pct_change().dropna()
    years = len(eq) / bars_per_year
    total_return = eq.iloc[-1] / eq.iloc[0] - 1.0
    cagr = (eq.iloc[-1] / eq.iloc[0]) ** (1 / years) - 1 if years > 0 else float("nan")

    std = ret.std(ddof=1)
    sharpe = (ret.mean() / std * np.sqrt(bars_per_year)) if std > 0 else float("nan")

    downside = ret[ret < 0]
    dstd = downside.std(ddof=1) if len(downside) > 1 else np.nan
    sortino = (ret.mean() / dstd * np.sqrt(bars_per_year)) if dstd and dstd > 0 else float("nan")

    mdd, trough, longest = max_drawdown(eq)
    calmar = cagr / abs(mdd) if mdd < 0 else float("nan")

    # トレード単位の統計
    fills = [f for f in result.fills if f.reason == "signal"]
    fee_total = result.stats.get("total_fee", 0.0)
    funding_total = result.stats.get("total_funding", 0.0)
    gross = eq.iloc[-1] - eq.iloc[0] + fee_total + funding_total

    return {
        "期間(年)": round(years, 2),
        "総リターン": f"{total_return:+.2%}",
        "CAGR": f"{cagr:+.2%}",
        "Sharpe": round(float(sharpe), 2),
        "Sortino": round(float(sortino), 2),
        "最大DD": f"{mdd:.2%}",
        "最大DD時刻": str(trough),
        "最長水没(本)": longest,
        "Calmar": round(float(calmar), 2),
        "約定回数": len(fills),
        "手数料合計": round(fee_total, 2),
        "資金調達合計": round(funding_total, 2),
        "コスト前損益": round(gross, 2),
        "コスト比率": (f"{(fee_total + funding_total) / abs(gross):.1%}"
                     if gross else "n/a"),
        "清算": "あり" if result.liquidated else "なし",
    }


def monthly_table(result: BacktestResult) -> pd.DataFrame:
    eq = result.equity.dropna()
    m = eq.resample("ME").last().pct_change().dropna()
    df = m.to_frame("return")
    df["year"] = df.index.year
    df["month"] = df.index.month
    return df.pivot_table(index="year", columns="month", values="return")


def print_report(result: BacktestResult, bars_per_year: int = 8_760) -> None:
    print("=" * 62)
    for k, v in metrics(result, bars_per_year).items():
        print(f"  {k:<14}: {v}")
    print("=" * 62)
    print("\n月次リターン")
    with pd.option_context("display.float_format", lambda v: f"{v:+.1%}"):
        print(monthly_table(result).to_string())
    print("\n※ 過去の結果は将来の成績を保証しません。"
          "この数値は手数料・スリッページ・資金調達を含む条件での試算です。")

"""run_backtest.py — 一連の流れをつなぐ実行スクリプト
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。
"""
from __future__ import annotations

import pandas as pd

from bt.engine import BacktestConfig, Backtester
from bt.fill import FillModel, PessimisticFillModel
from bt.report import print_report
from bt.strategy import SmaCrossStrategy
from client import make_client
from costs import FeeModel, fetch_fee_model, fetch_funding_history, funding_interval_hours
from kline_store import KlineStore


def main() -> None:
    c = make_client()
    store = KlineStore()

    SYMBOL, INTERVAL = "BTCUSDT", "60"
    bars = store.load(SYMBOL, INTERVAL)
    if bars.empty:
        raise SystemExit("先に記事07のKlineStore.update()でデータを取得してください")
    bars = bars.loc["2024-01-01":]

    # 実際の手数料と資金調達率を使う
    try:
        fees = fetch_fee_model(c.http, SYMBOL)
    except Exception as e:
        print(f"手数料取得に失敗（{e}）。既定値を使います")
        from decimal import Decimal
        fees = FeeModel(maker=Decimal("0.0002"), taker=Decimal("0.00055"))

    start_ms = int(bars.index[0].timestamp() * 1000)
    funding = fetch_funding_history(c.http, SYMBOL, start_ms)["funding_rate"]
    hours = funding_interval_hours(c.http, SYMBOL)

    cfg = BacktestConfig(initial_cash=10_000, funding_interval_hours=hours)

    for label, model in (("標準モデル", FillModel(slippage_bps=2.0, maker_fill_prob=0.7)),
                         ("悲観モデル", PessimisticFillModel())):
        print(f"\n########## {label} ##########")
        bt = Backtester(bars, fees, fill_model=model, funding=funding, config=cfg)
        result = bt.run(SmaCrossStrategy(fast=20, slow=60, use_post_only=True))
        print_report(result, bars_per_year=8_760)


if __name__ == "__main__":
    main()

"""test_lookahead.py — 未来参照の自動検出
本コードは本人利用に限り自由に使用可。再配布・商用利用・販売は禁止。

原理: ある時点t以降のデータを壊しても、t以前の判断が1つも変わらなければ、
      その戦略はtの時点で未来を見ていない。
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from bt.engine import Backtester
from bt.strategy import Strategy


def orders_signature(bt: Backtester, strategy: Strategy) -> list[tuple]:
    """発注の履歴を、比較可能なタプル列として取り出す。"""
    result = bt.run(strategy)
    return [(f.ts, f.side.value, round(f.qty, 8), f.tag) for f in result.fills]


def assert_no_lookahead(bars: pd.DataFrame, fees, make_strategy,
                        cut_ratio: float = 0.7, seed: int = 0) -> None:
    """未来を壊した場合と壊さない場合で、cut以前の発注が一致するか検証する。"""
    cut = int(len(bars) * cut_ratio)
    cut_ts = bars.index[cut]

    base = orders_signature(Backtester(bars, fees), make_strategy())

    # cut以降の価格をランダムに壊す（未来だけを別物にする）
    rng = np.random.default_rng(seed)
    tampered = bars.copy()
    idx = tampered.index[cut:]
    noise = rng.uniform(0.5, 1.5, size=len(idx))
    for col in ("open", "high", "low", "close"):
        tampered.loc[idx, col] = tampered.loc[idx, col].to_numpy() * noise
    # high/lowの整合性を保つ
    tampered.loc[idx, "high"] = tampered.loc[idx, ["open", "high", "low", "close"]].max(axis=1)
    tampered.loc[idx, "low"] = tampered.loc[idx, ["open", "high", "low", "close"]].min(axis=1)

    tampered_sig = orders_signature(Backtester(tampered, fees), make_strategy())

    before_base = [o for o in base if o[0] <= cut_ts]
    before_tamp = [o for o in tampered_sig if o[0] <= cut_ts]

    if before_base != before_tamp:
        # 最初に食い違った箇所を報告する
        for i, (a, b) in enumerate(zip(before_base, before_tamp)):
            if a != b:
                raise AssertionError(
                    f"ルックアヘッドを検出: {i}番目の発注が未来の改変で変化しました\n"
                    f"  元      : {a}\n  改変後  : {b}"
                )
        raise AssertionError(
            f"ルックアヘッドを検出: 発注件数が変化しました "
            f"({len(before_base)} -> {len(before_tamp)})"
        )
    print(f"[OK] ルックアヘッドなし（cut={cut_ts}, "
          f"検証対象 {len(before_base)}件の発注）")


if __name__ == "__main__":
    from decimal import Decimal

    from bt.strategy import SmaCrossStrategy
    from costs import FeeModel
    from kline_store import KlineStore

    bars = KlineStore().load("BTCUSDT", "60")
    fees = FeeModel(maker=Decimal("0.0002"), taker=Decimal("0.00055"))

    # 健全な戦略 -> 通る
    assert_no_lookahead(bars, fees, lambda: SmaCrossStrategy(20, 60))

    # わざと未来を見る戦略 -> 検出されるべき
    class Cheater(SmaCrossStrategy):
        def on_bar(self, history, position, equity):
            # historyには未来が入っていないので、本来はこれで漏れない。
            # 実際の事故は「エンジン外で全期間のindicatorを事前計算し、
            # その値を参照する」形で起きる。下はそれを模したもの。
            if getattr(self, "_precomputed", None) is None:
                self._precomputed = bars["close"].shift(-1)   # 未来の終値
            fut = self._precomputed.get(history.index[-1])
            if fut is None or pd.isna(fut):
                return None
            from bt.types import Order, OrderType, Side
            side = Side.BUY if fut > history["close"].iloc[-1] else Side.SELL
            if position.side and position.side is not side:
                return [Order(side=side, qty=abs(position.qty),
                              order_type=OrderType.MARKET, reduce_only=True,
                              tag="cheat-exit")]
            if position.is_flat:
                qty = round(equity * 0.5 / history["close"].iloc[-1], 6)
                return [Order(side=side, qty=qty, tag="cheat")]
            return None

    try:
        assert_no_lookahead(bars, fees, Cheater)
        print("[NG] 検出できませんでした（テストが機能していません）")
    except AssertionError as e:
        print(f"[OK] 期待どおり検出しました:\n{e}")