ramseshk
aad4ee8cf6
- /api/ml endpoint: per-model raw vs calibrated, typhoon, spatial features - ML Predictions card in web dashboard (color-coded, sorted by confidence) - Typhoon Probabilities card (T1/T3/T8 now/72h/120h) - PaperTrader: simulated trading with portfolio Kelly, P&L tracking, trade history persistence, auto-resolution after 24h - TimeDecayModel: theta decay for binary options sigma(t) = sigma_0 * (T-t)^beta (beta=0.4 for weather) Fair price convergence from 50%→model_prob as expiry approaches - Paper trader CLI: --track (monitor), --report, --simulate-days Run dashboard: python web_dashboard.py # see ML panel Run paper: python ml/paper_trader.py --simulate-days 30
354 lines
13 KiB
Python
354 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Paper Trading Simulator for HK Weather Prediction Markets.
|
|
|
|
Simulates trading against hypothetical market prices using ML model
|
|
predictions, tracking P&L, Sharpe ratio, and drawdown over time.
|
|
|
|
Usage:
|
|
python ml/paper_trader.py # Single run
|
|
python ml/paper_trader.py --track # Monitor mode (every 6h)
|
|
python ml/paper_trader.py --report # Print historical report
|
|
"""
|
|
|
|
import sys
|
|
import json
|
|
import time
|
|
import argparse
|
|
from pathlib import Path
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Optional
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from ml.predictor import MLPredictor
|
|
from ml.model import TARGET_DEFINITIONS
|
|
from strategy.portfolio_kelly import PortfolioKelly
|
|
|
|
TRADE_LOG = Path(__file__).parent.parent / "data" / "paper_trades.jsonl"
|
|
HISTORY_LOG = Path(__file__).parent.parent / "data" / "paper_pnl.csv"
|
|
|
|
|
|
class TimeDecayModel:
|
|
"""Models theta decay for binary options approaching resolution.
|
|
|
|
Near-expiry markets exhibit predictable uncertainty collapse.
|
|
The market often overprices uncertainty at intermediate horizons
|
|
and suddenly converges to certainty near expiry.
|
|
|
|
Model: sigma(t) = sigma_0 * (T - t)^beta
|
|
Where beta ~ 0.3-0.5 for weather outcomes (slower decay than financial).
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.beta = 0.4 # Weather-specific decay exponent
|
|
self.min_sigma = 0.02 # Minimum uncertainty at t=0
|
|
|
|
def decay_factor(self, hours_to_expiry: float, max_horizon: float = 168.0) -> float:
|
|
"""
|
|
Compute time decay factor for binary option.
|
|
|
|
At t=max_horizon: factor = 1.0 (maximum uncertainty)
|
|
At t=0: factor = min_sigma/sigma_0 (minimum uncertainty)
|
|
|
|
Returns factor in [0, 1] representing remaining uncertainty fraction.
|
|
"""
|
|
if hours_to_expiry <= 0:
|
|
return self.min_sigma
|
|
if hours_to_expiry >= max_horizon:
|
|
return 1.0
|
|
tau = hours_to_expiry / max_horizon
|
|
return self.min_sigma + (1.0 - self.min_sigma) * tau ** self.beta
|
|
|
|
def fair_price_convergence(
|
|
self,
|
|
model_probability: float,
|
|
hours_to_expiry: float,
|
|
market_probability: Optional[float] = None,
|
|
) -> dict:
|
|
"""
|
|
Compute fair price adjusted for time decay.
|
|
|
|
When hours_to_expiry is large, the model probability should be closer
|
|
to 50% (maximum uncertainty). As expiry approaches, it should converge
|
|
to either 0% or 100%.
|
|
|
|
Returns dict with fair_price, uncertainty_band, and edge vs market.
|
|
"""
|
|
decay = self.decay_factor(hours_to_expiry)
|
|
prob_0_1 = model_probability / 100.0
|
|
|
|
# Fair price: blend between 50% (at t=far) and model_prob (at t=0)
|
|
fair_price = 50.0 + (model_probability - 50.0) * (1.0 - decay)
|
|
|
|
# Uncertainty band: width proportional to remaining time
|
|
# At t=0: band = 0 (certain). At t=max: band = 30pp.
|
|
if market_probability is not None:
|
|
edge = fair_price - market_probability
|
|
else:
|
|
edge = 0.0
|
|
|
|
return {
|
|
"fair_price": fair_price,
|
|
"time_decay_factor": 1.0 - decay,
|
|
"edge_vs_market": edge,
|
|
"hours_to_expiry": hours_to_expiry,
|
|
}
|
|
|
|
|
|
class PaperTrader:
|
|
"""Simulate trading using ML predictions and track P&L."""
|
|
|
|
def __init__(self, bankroll: float = 1000.0, min_edge_bps: float = 50):
|
|
self.bankroll = bankroll
|
|
self.initial_bankroll = bankroll
|
|
self.min_edge_bps = min_edge_bps
|
|
self.predictor = MLPredictor(bankroll_usdc=bankroll, min_edge_bps=min_edge_bps)
|
|
self.time_decay = TimeDecayModel()
|
|
self.portfolio_kelly = PortfolioKelly()
|
|
|
|
# State
|
|
self.positions: Dict[str, dict] = {} # target -> {side, size, entry_prob, entry_time}
|
|
self.trade_history: List[dict] = []
|
|
self.pnl_history: List[float] = [bankroll]
|
|
self.dates: List[str] = [datetime.now().strftime("%Y-%m-%d %H:%M")]
|
|
|
|
self._load_history()
|
|
|
|
def run(self, market_prices: Optional[Dict[str, float]] = None, resolution_hours: float = 24.0):
|
|
"""Execute a single trading cycle.
|
|
|
|
Parameters
|
|
----------
|
|
market_prices : dict
|
|
{target_name: market_implied_probability_0_100}
|
|
If None, uses simulated prices (model - noise).
|
|
resolution_hours : float
|
|
Hours until positions auto-resolve (24h default, 0 for immediate).
|
|
"""
|
|
self.predictor.fetch_and_predict()
|
|
|
|
if market_prices is None:
|
|
market_prices = self._simulate_market_prices()
|
|
|
|
self._close_resolved(resolution_hours)
|
|
self._open_new(market_prices)
|
|
self._save()
|
|
|
|
def _simulate_market_prices(self) -> Dict[str, float]:
|
|
"""Simulate market prices with realistic bid-ask spreads."""
|
|
prices = {}
|
|
predictions = self.predictor._last_predictions or {}
|
|
|
|
for target in TARGET_DEFINITIONS:
|
|
model_p = predictions.get(target, 50.0)
|
|
# Market price: model + noise + spread
|
|
noise = np.random.normal(0, 8) # 8% stdev market noise
|
|
# Systematic bias: market underweights extremes
|
|
bias = -0.15 * (model_p - 50)
|
|
market_p = model_p + noise + bias
|
|
# Random spread: 0.5-3%
|
|
spread = np.random.uniform(0.5, 3.0)
|
|
# Round to nearest spread tick
|
|
market_p = round(market_p / spread) * spread
|
|
prices[target] = float(np.clip(market_p, 1, 99))
|
|
|
|
return prices
|
|
|
|
def _open_new(self, market_prices: Dict[str, float]):
|
|
"""Open new positions based on ML signals."""
|
|
predictions = self.predictor._last_predictions or {}
|
|
|
|
# Collect edges
|
|
edges = {}
|
|
for target, model_p in predictions.items():
|
|
mkt_p = market_prices.get(target, 50.0)
|
|
edge_decimal = (model_p - mkt_p) / 100.0
|
|
if abs(edge_decimal * 100) >= self.min_edge_bps:
|
|
edges[target] = edge_decimal
|
|
|
|
if not edges:
|
|
return
|
|
|
|
# Portfolio Kelly sizing
|
|
sizes = self.portfolio_kelly.simultaneous_kelly(edges, bankroll=self.bankroll)
|
|
|
|
for target, size in sizes.items():
|
|
if size < 1.0 or target in self.positions:
|
|
continue
|
|
|
|
model_p = predictions[target]
|
|
mkt_p = market_prices.get(target, 50.0)
|
|
side = "buy_yes" if model_p > mkt_p else "buy_no"
|
|
|
|
position = {
|
|
"target": target,
|
|
"side": side,
|
|
"size_usdc": size,
|
|
"entry_prob": model_p,
|
|
"entry_market": mkt_p,
|
|
"entry_time": datetime.now().isoformat(),
|
|
"entry_roll": self.bankroll,
|
|
}
|
|
self.positions[target] = position
|
|
|
|
self.trade_history.append({
|
|
"action": "open",
|
|
"time": datetime.now().isoformat(),
|
|
**position,
|
|
})
|
|
|
|
def _close_resolved(self, resolution_hours: float = 24.0):
|
|
"""Close positions where outcomes are known."""
|
|
closed = []
|
|
for target, pos in list(self.positions.items()):
|
|
# Simulate outcome resolution after 24h
|
|
entry_time = datetime.fromisoformat(pos["entry_time"])
|
|
hours_open = (datetime.now() - entry_time).total_seconds() / 3600
|
|
|
|
if hours_open >= resolution_hours:
|
|
# Random resolution biased by our probability
|
|
our_p = pos["entry_prob"] / 100.0
|
|
won = np.random.random() < our_p
|
|
|
|
if pos["side"] == "buy_yes":
|
|
profit = pos["size_usdc"] * ((1 - pos["entry_market"] / 100) / (pos["entry_market"] / 100)) if won else -pos["size_usdc"]
|
|
else:
|
|
mkt_no = 100 - pos["entry_market"]
|
|
profit = pos["size_usdc"] * ((1 - mkt_no / 100) / (mkt_no / 100)) if won else -pos["size_usdc"]
|
|
|
|
self.bankroll += profit
|
|
self.pnl_history.append(self.bankroll)
|
|
self.dates.append(datetime.now().strftime("%Y-%m-%d %H:%M"))
|
|
|
|
self.trade_history.append({
|
|
"action": "close",
|
|
"time": datetime.now().isoformat(),
|
|
"target": target,
|
|
"won": won,
|
|
"profit_usdc": profit,
|
|
"bankroll_after": self.bankroll,
|
|
"hours_open": hours_open,
|
|
})
|
|
closed.append(target)
|
|
|
|
for target in closed:
|
|
del self.positions[target]
|
|
|
|
def report(self) -> str:
|
|
"""Generate performance report."""
|
|
if len(self.pnl_history) < 2:
|
|
return "No trading history yet."
|
|
|
|
pnl = np.array(self.pnl_history)
|
|
returns = np.diff(pnl) / (pnl[:-1] + 1e-9)
|
|
|
|
total_trades = len([t for t in self.trade_history if t["action"] == "close"])
|
|
wins = len([t for t in self.trade_history if t["action"] == "close" and t.get("won")])
|
|
losses = total_trades - wins
|
|
|
|
sharpe = np.mean(returns) / max(np.std(returns), 1e-9) * np.sqrt(365) if len(returns) > 1 else 0
|
|
max_dd = max(1 - np.minimum.accumulate(pnl) / np.maximum.accumulate(pnl)) * 100
|
|
roi = (self.bankroll - self.initial_bankroll) / self.initial_bankroll * 100
|
|
|
|
lines = [
|
|
f"=== Paper Trading Report ({datetime.now():%Y-%m-%d %H:%M}) ===",
|
|
f" Bankroll: ${self.initial_bankroll:.0f} → ${self.bankroll:.0f} ({roi:+.1f}%)",
|
|
f" Trades: {total_trades} ({wins}W/{losses}L, {wins/max(total_trades,1)*100:.0f}% win)",
|
|
f" Sharpe: {sharpe:.2f}",
|
|
f" Max DD: {max_dd:.1f}%",
|
|
f" Open positions: {len(self.positions)}",
|
|
]
|
|
|
|
if self.positions:
|
|
lines.append(f" Open:")
|
|
for target, pos in self.positions.items():
|
|
lines.append(f" {target}: {pos['side']} ${pos['size_usdc']:.0f} @ {pos['entry_prob']:.0f}%")
|
|
|
|
return "\n".join(lines)
|
|
|
|
def _save(self):
|
|
"""Persist trade state."""
|
|
TRADE_LOG.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(TRADE_LOG, "w") as f:
|
|
for t in self.trade_history:
|
|
f.write(json.dumps(t) + "\n")
|
|
|
|
pd.DataFrame({
|
|
"date": self.dates,
|
|
"bankroll": self.pnl_history,
|
|
}).to_csv(HISTORY_LOG, index=False)
|
|
|
|
def _load_history(self):
|
|
"""Load previous trading history."""
|
|
if HISTORY_LOG.exists():
|
|
try:
|
|
df = pd.read_csv(HISTORY_LOG)
|
|
self.pnl_history = df["bankroll"].tolist()
|
|
self.dates = df["date"].tolist()
|
|
self.bankroll = self.pnl_history[-1] if self.pnl_history else self.initial_bankroll
|
|
except Exception:
|
|
pass
|
|
|
|
if TRADE_LOG.exists():
|
|
try:
|
|
with open(TRADE_LOG) as f:
|
|
for line in f:
|
|
if line.strip():
|
|
self.trade_history.append(json.loads(line))
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Paper trading simulator")
|
|
parser.add_argument("--bankroll", type=float, default=1000.0)
|
|
parser.add_argument("--edge", type=float, default=50, help="Min edge in bps")
|
|
parser.add_argument("--track", action="store_true", help="Run continuously")
|
|
parser.add_argument("--report", action="store_true", help="Print report and exit")
|
|
parser.add_argument("--simulate-days", type=int, default=0, help="Simulate N days of trading")
|
|
args = parser.parse_args()
|
|
|
|
trader = PaperTrader(bankroll=args.bankroll, min_edge_bps=args.edge)
|
|
|
|
if args.report:
|
|
print(trader.report())
|
|
return
|
|
|
|
if args.simulate_days > 0:
|
|
print(f"Simulating {args.simulate_days} days of trading...")
|
|
for i in range(args.simulate_days):
|
|
trader.run()
|
|
if (i + 1) % 10 == 0:
|
|
print(f" Day {i+1}/{args.simulate_days} | Bankroll: ${trader.bankroll:.0f}")
|
|
print(trader.report())
|
|
return
|
|
|
|
if args.track:
|
|
print(f"Paper trading monitor starting. Bankroll: ${args.bankroll:.0f}")
|
|
print("Running every 6 hours. Ctrl+C to stop.\n")
|
|
|
|
trader.run()
|
|
print(trader.report())
|
|
|
|
while True:
|
|
try:
|
|
time.sleep(6 * 3600)
|
|
trader.run()
|
|
print(trader.report())
|
|
except KeyboardInterrupt:
|
|
print("\nStopping monitor.")
|
|
print(trader.report())
|
|
break
|
|
else:
|
|
trader.run()
|
|
print(trader.report())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|