89 lines
3.8 KiB
Python
89 lines
3.8 KiB
Python
"""
|
|
Day-by-day replay of a fixed set of REAL legs (exact expiry/strike from a real Saxo chain,
|
|
built the normal Strategy Builder way) against the ACTUALLY accumulated Saxo history —
|
|
not a hypothetical scenario, a mark-to-market of what really happened between two dates
|
|
that are both within services.option_chain's accumulated snapshot depth (currently up to
|
|
~120 days — see services.saxo_client.snapshot_options_chain's max_days).
|
|
|
|
This answers a different question than Strategy Builder's own scenario pricing ("what
|
|
would this be worth if spot moved X% and IV moved Y%") — here nothing is guessed, every
|
|
day's mark comes from a real quote captured that day, or the position isn't valued for a
|
|
day where any leg has no real quote (skipped, not synthesized — a replay should show what
|
|
was actually knowable, not fill gaps with a theoretical price).
|
|
"""
|
|
from datetime import date, timedelta
|
|
from typing import Any, Dict, List
|
|
|
|
|
|
def _daterange(start_date: str, end_date: str) -> List[str]:
|
|
d0 = date.fromisoformat(start_date[:10])
|
|
d1 = date.fromisoformat(end_date[:10])
|
|
return [(d0 + timedelta(days=i)).isoformat() for i in range((d1 - d0).days + 1)]
|
|
|
|
|
|
def replay_position(
|
|
symbol: str, legs: List[Dict[str, Any]], start_date: str, end_date: str,
|
|
contract_size: float = 100_000,
|
|
) -> Dict[str, Any]:
|
|
from services.database import get_saxo_option_symbol_for_ticker
|
|
from services.option_chain import get_chain_slice, find_quote
|
|
|
|
if end_date <= start_date:
|
|
raise ValueError("La date de fin doit être postérieure à la date de départ.")
|
|
if not legs:
|
|
raise ValueError("Aucune jambe à rejouer.")
|
|
|
|
saxo_symbol = get_saxo_option_symbol_for_ticker(symbol) or symbol.upper()
|
|
|
|
signed_qty = [(1 if leg["position"] == "long" else -1) * leg.get("quantity", 1) for leg in legs]
|
|
avg_days = sum(leg.get("days_to_expiry", 30) for leg in legs) / len(legs)
|
|
|
|
points: List[Dict[str, Any]] = []
|
|
entry_value = None
|
|
missing_dates: List[str] = []
|
|
|
|
for d in _daterange(start_date, end_date):
|
|
try:
|
|
# n_expiries wide enough to virtually guarantee every expiry the legs use is
|
|
# present regardless of how target_days ranks them from this day's viewpoint —
|
|
# accumulated history rarely holds more than ~20 distinct expiries per symbol.
|
|
chain = get_chain_slice(saxo_symbol, target_days=int(avg_days), n_expiries=25, dte_min=0, dte_max=400, as_of=d)
|
|
except ValueError:
|
|
missing_dates.append(d)
|
|
continue
|
|
|
|
value = 0.0 # dollar value of the whole position, contract_size already applied
|
|
complete = True
|
|
for leg, sq in zip(legs, signed_qty):
|
|
q = find_quote(chain, leg["expiry_date"], leg["strike"], leg["option_type"])
|
|
if not q or q["mid"] <= 0:
|
|
complete = False
|
|
break
|
|
value += sq * q["mid"] * contract_size
|
|
if not complete:
|
|
missing_dates.append(d)
|
|
continue
|
|
|
|
if entry_value is None:
|
|
entry_value = value
|
|
points.append({
|
|
"date": d, "spot": chain.get("spot"),
|
|
"position_value": round(value, 2),
|
|
"pnl": round(value - entry_value, 2),
|
|
})
|
|
|
|
if not points:
|
|
raise ValueError(
|
|
f"Aucune cotation réelle exploitable pour ces jambes entre {start_date} et {end_date} "
|
|
"— vérifiez que ces strikes/échéances exactes ont bien été cotés par Saxo sur cette période."
|
|
)
|
|
|
|
return {
|
|
"symbol": symbol, "saxo_symbol": saxo_symbol,
|
|
"start_date": start_date, "end_date": end_date,
|
|
"entry_date": points[0]["date"], "entry_value": round(entry_value, 2),
|
|
"final_pnl": points[-1]["pnl"],
|
|
"points": points,
|
|
"missing_dates": missing_dates,
|
|
}
|