95 lines
3.5 KiB
Python
95 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Dict
|
|
|
|
from .dsl import PlanDelta
|
|
|
|
|
|
class Backtester:
|
|
"""Toy deterministic replay-based backtester for MVP.
|
|
|
|
Exposes an apply() method that consumes a Signals stream and a PlanDelta
|
|
to produce a final cash amount, suitable for the tests in this repo.
|
|
Also provides a lightweight replay() helper used by tests.
|
|
"""
|
|
|
|
def __init__(self, seed=None, initial_cash: float = 0.0):
|
|
self.seed = seed
|
|
self.initial_cash = initial_cash
|
|
|
|
def run(self, plan: PlanDelta) -> Dict[str, float]:
|
|
# Backwards-compatible helper using the same simple cost model as apply()
|
|
def _entries(p):
|
|
if p is None:
|
|
return []
|
|
if hasattr(p, "deltas") and p.deltas:
|
|
return p.deltas
|
|
if hasattr(p, "delta") and p.delta:
|
|
return p.delta
|
|
return []
|
|
|
|
entries = _entries(plan)
|
|
hedge_count = len(entries)
|
|
total_cost = 0.0
|
|
for entry in entries:
|
|
if isinstance(entry, dict):
|
|
size = abs(float(entry.get("size", 0.0)))
|
|
price = float(entry.get("price", 0.0))
|
|
else:
|
|
size = getattr(entry, "size", 0.0)
|
|
price = getattr(entry, "price", 0.0)
|
|
total_cost += size * price
|
|
pnl = max(0.0, 0.0 - total_cost) # placeholder deterministic path
|
|
return {"deterministic_pnl": pnl, "hedge_count": hedge_count}
|
|
|
|
def replay(self, signals, plan: PlanDelta) -> float:
|
|
"""Deterministic replay API used by tests.
|
|
Returns a float PnL placeholder based on plan size.
|
|
"""
|
|
total_cost = 0.0
|
|
def _iter_entries(p):
|
|
if not p:
|
|
return []
|
|
if hasattr(p, "deltas") and p.deltas:
|
|
return p.deltas
|
|
if hasattr(p, "delta") and p.delta:
|
|
return p.delta
|
|
return []
|
|
|
|
for entry in _iter_entries(plan):
|
|
if isinstance(entry, dict):
|
|
total_cost += abs(float(entry.get("size", 0.0))) * float(entry.get("price", 0.0))
|
|
else:
|
|
# Try common attribute-based access for StrategyDelta-like objects
|
|
size = getattr(entry, "size", 0.0)
|
|
price = getattr(entry, "price", 0.0)
|
|
if size is not None and price is not None:
|
|
total_cost += abs(float(size)) * float(price)
|
|
# Simple deterministic path: final cash is initial minus total_cost
|
|
return float(self.initial_cash) - total_cost
|
|
|
|
def apply(self, signals, plan: PlanDelta) -> float:
|
|
"""Apply a sequence of MarketSignals against a PlanDelta to compute final cash.
|
|
Cost is modeled as sum(|size| * price) for each hedge-like action in plan.delta.
|
|
Final cash = initial_cash - total_cost.
|
|
"""
|
|
total_cost = 0.0
|
|
def _entries(p):
|
|
if p is None:
|
|
return []
|
|
if hasattr(p, "deltas") and p.deltas:
|
|
return p.deltas
|
|
if hasattr(p, "delta") and p.delta:
|
|
return p.delta
|
|
return []
|
|
for entry in _entries(plan):
|
|
if isinstance(entry, dict):
|
|
size = abs(float(entry.get("size", 0.0)))
|
|
price = float(entry.get("price", 0.0))
|
|
else:
|
|
size = getattr(entry, "size", 0.0)
|
|
price = getattr(entry, "price", 0.0)
|
|
total_cost += size * price
|
|
final_cash = float(self.initial_cash) - total_cost
|
|
return final_cash
|