43 lines
1.6 KiB
Python
43 lines
1.6 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.
|
|
"""
|
|
|
|
def __init__(self, initial_cash: float = 0.0):
|
|
self.initial_cash = initial_cash
|
|
|
|
def run(self, plan: PlanDelta) -> Dict[str, float]:
|
|
# Backwards-compatible helper using the same simple cost model as apply()
|
|
hedge_count = len(plan.delta) if plan and plan.delta else 0
|
|
total_cost = 0.0
|
|
if plan and plan.delta:
|
|
for entry in plan.delta:
|
|
size = abs(float(entry.get("size", 0.0)))
|
|
price = float(entry.get("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 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
|
|
if plan and plan.delta:
|
|
for entry in plan.delta:
|
|
size = abs(float(entry.get("size", 0.0)))
|
|
price = float(entry.get("price", 0.0))
|
|
total_cost += size * price
|
|
final_cash = float(self.initial_cash) - total_cost
|
|
return final_cash
|