polymarket-crypto-toolkit/backtest_engine.py at main · 0xrsydn/polymarket-crypto-toolkit · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# DEPRECATED: Use polymarket_algo.* packages instead. This file exists for backward compatibility.
from __future__ import annotations

from collections.abc import Callable
from dataclasses import dataclass
from itertools import product
from typing import Any

import numpy as np
import pandas as pd

StrategyCallable = Callable[..., pd.Series | pd.DataFrame]


@dataclass
class BacktestResult:
    metrics: dict[str, Any]
    trades: pd.DataFrame
    pnl_curve: pd.Series


def _max_drawdown(equity_curve: pd.Series) -> float:
    running_max = equity_curve.cummax()
    drawdown = equity_curve - running_max
    return float(drawdown.min()) if not drawdown.empty else 0.0


def run_backtest(
    candles: pd.DataFrame,
    strategy: StrategyCallable,
    strategy_params: dict[str, Any] | None = None,
    buy_price: float = 0.50,
    win_payout: float = 0.95,
) -> BacktestResult:
    strategy_params = strategy_params or {}

    out = strategy(candles, **strategy_params)
    if isinstance(out, pd.DataFrame):
        signals = out["signal"].astype(int)
        size = out.get("size", pd.Series(15.0, index=candles.index)).astype(float)
    else:
        signals = out.astype(int)
        size = pd.Series(15.0, index=candles.index)

    next_close = candles["close"].shift(-1)
    outcome_up = (next_close > candles["close"]).astype(int)

    active = (signals != 0) & outcome_up.notna()
    direction_up = signals == 1
    wins = (direction_up & (outcome_up == 1)) | ((signals == -1) & (outcome_up == 0))

    per_share_pnl = np.where(wins, win_payout - buy_price, -buy_price)
    per_share_pnl = pd.Series(per_share_pnl, index=candles.index)

    trade_pnl = (per_share_pnl * size).where(active, 0.0)
    pnl_curve = trade_pnl.cumsum()

    trades = pd.DataFrame(
        {
            "timestamp": candles.index,
            "signal": signals,
            "size": size,
            "entry_close": candles["close"],
            "next_close": next_close,
            "is_win": wins.where(active, False),
            "pnl": trade_pnl,
        }
    )
    trades = trades.loc[active]

    trade_count = int(active.sum())
    win_rate = float(trades["is_win"].mean()) if trade_count else 0.0
    total_pnl = float(trade_pnl.sum())
    returns = trade_pnl.loc[active]
    sharpe = (
        float((returns.mean() / returns.std(ddof=0)) * np.sqrt(len(returns)))
        if trade_count and returns.std(ddof=0) > 0
        else 0.0
    )

    metrics = {
        "win_rate": win_rate,
        "total_pnl": total_pnl,
        "max_drawdown": _max_drawdown(pnl_curve),
        "sharpe_ratio": sharpe,
        "trade_count": trade_count,
    }

    return BacktestResult(metrics=metrics, trades=trades, pnl_curve=pnl_curve)


def parameter_sweep(
    candles: pd.DataFrame,
    strategy: StrategyCallable,
    param_grid: dict[str, list[Any]],
) -> pd.DataFrame:
    keys = list(param_grid.keys())
    rows: list[dict[str, Any]] = []

    for values in product(*[param_grid[k] for k in keys]):
        params = dict(zip(keys, values, strict=False))
        result = run_backtest(candles, strategy, params)
        rows.append({**params, **result.metrics})

    return pd.DataFrame(rows).sort_values(by=["win_rate", "total_pnl"], ascending=False).reset_index(drop=True)


def walk_forward_split(candles: pd.DataFrame, train_ratio: float = 0.75) -> tuple[pd.DataFrame, pd.DataFrame]:
    split_idx = int(len(candles) * train_ratio)
    train = candles.iloc[:split_idx].copy()
    test = candles.iloc[split_idx:].copy()
    return train, test