feat: 3-tier outcome scoring + options P&L simulation in Pattern Lab
Backend (pattern_lab.py): - Replace binary HIT/MISS with FULL / PARTIAL / MISS scoring FULL: right direction AND ≥ 50% of expected move PARTIAL: right direction AND ≥ 15% of expected move (was always MISS before) MISS: wrong direction or negligible move - Add direction_correct, direction_ratio, hit_type fields to all outcomes - Add Black-Scholes ATM options P&L simulation (_bs_price, _ncdf, _sigma_for) Normalised to S₀=K=100, per-asset-class vol heuristic (FX 8%, indices 16%, crypto 65%) Supports: long call/put, straddle, strangle, call spread, put spread - estimated_options_pnl_pct shows what the strategy would have returned Frontend (PatternLab.tsx): - OutcomeRow component: FULL HIT (green) / PARTIAL (amber) / MISS (red) - Shows direction tick/cross + ratio % of target achieved - Shows estimated options P&L with DollarSign icon - Hit rate header shows full hits + partial count separately - Card border: emerald = full, amber = partial, red = miss Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OpenSquared
@@ -9,6 +9,7 @@ Workflow:
|
||||
import json
|
||||
import logging
|
||||
from datetime import datetime, timedelta
|
||||
from math import log, sqrt, exp, erf
|
||||
from typing import Optional
|
||||
|
||||
import numpy as np
|
||||
@@ -18,6 +19,140 @@ import yfinance as yf
|
||||
_log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ── Options P&L helpers ────────────────────────────────────────────────────────
|
||||
|
||||
def _ncdf(x: float) -> float:
|
||||
return 0.5 * (1.0 + erf(x / sqrt(2.0)))
|
||||
|
||||
|
||||
def _bs_price(S: float, K: float, T: float, r: float, sigma: float, opt_type: str) -> float:
|
||||
"""Black-Scholes European option price. T in years. At expiry (T≤0) returns intrinsic."""
|
||||
if T <= 0:
|
||||
return max(S - K, 0.0) if opt_type == "call" else max(K - S, 0.0)
|
||||
d1 = (log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * sqrt(T))
|
||||
d2 = d1 - sigma * sqrt(T)
|
||||
if opt_type == "call":
|
||||
return S * _ncdf(d1) - K * exp(-r * T) * _ncdf(d2)
|
||||
return K * exp(-r * T) * _ncdf(-d2) - S * _ncdf(-d1)
|
||||
|
||||
|
||||
def _sigma_for(ticker: str) -> float:
|
||||
"""Rough annualised vol estimate by asset class (ATM premium sizing)."""
|
||||
t = ticker.upper()
|
||||
if t.endswith("=X"): return 0.08 # FX pairs
|
||||
if "VIX" in t: return 0.80 # volatility index
|
||||
if t.startswith("^"): return 0.16 # equity indices
|
||||
if t.endswith("-USD") or "-USD" in t: return 0.65 # crypto
|
||||
if t.endswith("=F"): return 0.25 # commodity futures
|
||||
return 0.20 # ETFs / default
|
||||
|
||||
|
||||
def _options_pnl_pct(
|
||||
strategy: str, ticker: str,
|
||||
actual_move_pct: float, horizon_days: int,
|
||||
expected_direction: str,
|
||||
) -> Optional[float]:
|
||||
"""
|
||||
Estimate % P&L on the suggested options strategy assuming ATM entry,
|
||||
full-horizon hold, and actual_move_pct underlying move by expiry.
|
||||
Normalised to S₀ = K = 100.
|
||||
"""
|
||||
try:
|
||||
sigma = _sigma_for(ticker)
|
||||
T = max(horizon_days / 365.0, 1 / 365.0)
|
||||
r = 0.03
|
||||
S0 = 100.0
|
||||
K = 100.0
|
||||
S1 = S0 * (1.0 + actual_move_pct / 100.0)
|
||||
strat = strategy.lower()
|
||||
|
||||
if "straddle" in strat:
|
||||
cost = _bs_price(S0, K, T, r, sigma, "call") + _bs_price(S0, K, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "call") + _bs_price(S1, K, 0, r, sigma, "put")
|
||||
elif "strangle" in strat:
|
||||
Kc, Kp = K * 1.05, K * 0.95
|
||||
cost = _bs_price(S0, Kc, T, r, sigma, "call") + _bs_price(S0, Kp, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, Kc, 0, r, sigma, "call") + _bs_price(S1, Kp, 0, r, sigma, "put")
|
||||
elif "call spread" in strat or "bull call" in strat:
|
||||
Kh = K * 1.10
|
||||
cost = _bs_price(S0, K, T, r, sigma, "call") - _bs_price(S0, Kh, T, r, sigma, "call")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "call") - _bs_price(S1, Kh, 0, r, sigma, "call")
|
||||
elif "put spread" in strat or "bear put" in strat:
|
||||
Kl = K * 0.90
|
||||
cost = _bs_price(S0, K, T, r, sigma, "put") - _bs_price(S0, Kl, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "put") - _bs_price(S1, Kl, 0, r, sigma, "put")
|
||||
elif "long call" in strat or ("call" in strat and "put" not in strat):
|
||||
cost = _bs_price(S0, K, T, r, sigma, "call")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "call")
|
||||
elif "long put" in strat or ("put" in strat and "call" not in strat):
|
||||
cost = _bs_price(S0, K, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "put")
|
||||
else:
|
||||
# Unknown strategy: fallback based on direction
|
||||
if expected_direction == "down":
|
||||
cost = _bs_price(S0, K, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "put")
|
||||
elif expected_direction == "up":
|
||||
cost = _bs_price(S0, K, T, r, sigma, "call")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "call")
|
||||
else:
|
||||
cost = _bs_price(S0, K, T, r, sigma, "call") + _bs_price(S0, K, T, r, sigma, "put")
|
||||
exit_v = _bs_price(S1, K, 0, r, sigma, "call") + _bs_price(S1, K, 0, r, sigma, "put")
|
||||
|
||||
if cost <= 0.01:
|
||||
return None
|
||||
return round((exit_v - cost) / cost * 100.0, 1)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _score_outcome(
|
||||
actual_move: float,
|
||||
expected_move: float,
|
||||
expected_dir: str,
|
||||
strategy: str,
|
||||
ticker: str,
|
||||
horizon_days: int,
|
||||
) -> dict:
|
||||
"""
|
||||
3-tier scoring:
|
||||
FULL — right direction AND ≥ 50% of expected magnitude
|
||||
PARTIAL — right direction AND ≥ 15% of expected magnitude
|
||||
MISS — wrong direction or negligible move
|
||||
`hit` (bool) = True only for FULL (for backtest_hits counter compatibility).
|
||||
"""
|
||||
em = abs(float(expected_move))
|
||||
|
||||
if expected_dir == "up":
|
||||
direction_correct = actual_move > 0
|
||||
direction_ratio = (actual_move / em) if em > 0 else 0.0
|
||||
elif expected_dir == "down":
|
||||
direction_correct = actual_move < 0
|
||||
direction_ratio = (-actual_move / em) if em > 0 else 0.0
|
||||
else: # "any" / volatility
|
||||
direction_correct = True
|
||||
direction_ratio = (abs(actual_move) / em) if em > 0 else 0.0
|
||||
|
||||
direction_ratio = round(direction_ratio, 3)
|
||||
|
||||
if direction_correct and direction_ratio >= 0.50:
|
||||
hit_type = "full"
|
||||
elif direction_correct and direction_ratio >= 0.15:
|
||||
hit_type = "partial"
|
||||
else:
|
||||
hit_type = "miss"
|
||||
|
||||
options_pnl = _options_pnl_pct(strategy, ticker, actual_move, horizon_days, expected_dir)
|
||||
|
||||
return {
|
||||
"hit": hit_type == "full",
|
||||
"hit_type": hit_type,
|
||||
"direction_correct": direction_correct,
|
||||
"direction_ratio": direction_ratio,
|
||||
"estimated_options_pnl_pct": options_pnl,
|
||||
}
|
||||
|
||||
|
||||
# ── Context builder ────────────────────────────────────────────────────────────
|
||||
|
||||
def _fetch_ticker(ticker: str, start: str, end: str) -> Optional[pd.DataFrame]:
|
||||
@@ -207,25 +342,20 @@ def evaluate_outcomes(run: dict) -> list:
|
||||
outcomes.append({"pattern_name": pat.get("name"), "underlying": ticker, "error": "insufficient history"})
|
||||
continue
|
||||
|
||||
entry_price = float(entry_sub["Close"].dropna().iloc[-1])
|
||||
exit_price = float(exit_sub["Close"].dropna().iloc[-1])
|
||||
actual_move = round((exit_price / entry_price - 1) * 100, 2)
|
||||
entry_price = float(entry_sub["Close"].dropna().iloc[-1])
|
||||
exit_price = float(exit_sub["Close"].dropna().iloc[-1])
|
||||
actual_move = round((exit_price / entry_price - 1) * 100, 2)
|
||||
|
||||
expected_dir = pat.get("expected_direction", "any")
|
||||
expected_move = float(pat.get("expected_move_pct", 0))
|
||||
threshold = max(expected_move * 0.5, 3.0) # at least 50% of expected, min 3%
|
||||
strategy = pat.get("strategy", "")
|
||||
|
||||
if expected_dir == "up":
|
||||
hit = actual_move >= threshold
|
||||
elif expected_dir == "down":
|
||||
hit = actual_move <= -threshold
|
||||
else: # any / volatility
|
||||
hit = abs(actual_move) >= threshold
|
||||
score = _score_outcome(actual_move, expected_move, expected_dir, strategy, ticker, horizon_days)
|
||||
|
||||
outcomes.append({
|
||||
"pattern_name": pat.get("name"),
|
||||
"underlying": ticker,
|
||||
"strategy": pat.get("strategy", ""),
|
||||
"strategy": strategy,
|
||||
"signal_direction": pat.get("signal_direction", ""),
|
||||
"expected_direction": expected_dir,
|
||||
"expected_move_pct": expected_move,
|
||||
@@ -234,8 +364,8 @@ def evaluate_outcomes(run: dict) -> list:
|
||||
"exit_price": round(exit_price, 4),
|
||||
"entry_date": analysis_date,
|
||||
"exit_date": eval_dt.strftime("%Y-%m-%d"),
|
||||
"hit": hit,
|
||||
"confidence": pat.get("confidence", 0),
|
||||
**score,
|
||||
})
|
||||
|
||||
return outcomes
|
||||
@@ -395,25 +525,20 @@ def evaluate_instrument_outcomes(run: dict) -> list:
|
||||
"analysis_date": pat_date, "error": "insufficient history"})
|
||||
continue
|
||||
|
||||
entry_price = float(entry_sub["Close"].dropna().iloc[-1])
|
||||
exit_price = float(exit_sub["Close"].dropna().iloc[-1])
|
||||
actual_move = round((exit_price / entry_price - 1) * 100, 2)
|
||||
entry_price = float(entry_sub["Close"].dropna().iloc[-1])
|
||||
exit_price = float(exit_sub["Close"].dropna().iloc[-1])
|
||||
actual_move = round((exit_price / entry_price - 1) * 100, 2)
|
||||
|
||||
expected_dir = pat.get("expected_direction", "any")
|
||||
expected_move = float(pat.get("expected_move_pct", 0))
|
||||
threshold = max(expected_move * 0.5, 3.0)
|
||||
strategy = pat.get("strategy", "")
|
||||
|
||||
if expected_dir == "up":
|
||||
hit = actual_move >= threshold
|
||||
elif expected_dir == "down":
|
||||
hit = actual_move <= -threshold
|
||||
else:
|
||||
hit = abs(actual_move) >= threshold
|
||||
score = _score_outcome(actual_move, expected_move, expected_dir, strategy, ticker, pat_horizon)
|
||||
|
||||
outcomes.append({
|
||||
"pattern_name": pat.get("name"),
|
||||
"underlying": ticker,
|
||||
"strategy": pat.get("strategy", ""),
|
||||
"strategy": strategy,
|
||||
"signal_direction": pat.get("signal_direction", ""),
|
||||
"expected_direction": expected_dir,
|
||||
"expected_move_pct": expected_move,
|
||||
@@ -423,8 +548,8 @@ def evaluate_instrument_outcomes(run: dict) -> list:
|
||||
"analysis_date": pat_date,
|
||||
"entry_date": pat_date,
|
||||
"exit_date": eval_dt.strftime("%Y-%m-%d"),
|
||||
"hit": hit,
|
||||
"confidence": pat.get("confidence", 0),
|
||||
**score,
|
||||
})
|
||||
|
||||
return outcomes
|
||||
|
||||
Reference in New Issue
Block a user