mahmoud20138 Skill: Risk And Portfolio | Domain: trading | Category: risk | Level: intermediate Tags:
trading,risk,portfolio,position-sizing,drawdown,kelly
Risk & Portfolio — Complete Framework
The safety layer. No trade should be taken without passing through risk checks.
Sections
- Risk Management — position sizing, rules, psychology, drawdown recovery
- Execution & Costs — entry timing, scaling, microstructure, spread/slippage
- Trade Journal & Performance — logging, analytics, equity curve
- Trade Filter — when NOT to trade (event blackouts, structural risks)
- Portfolio & Allocation — Markowitz, risk parity, baskets, strategy allocation
- Multi-Account Manager — aggregate P&L across brokers
- Tail Risk Hedging — black swan protection
- Risk Premia — factor investing (momentum, carry, value, volatility premium)
- Arbitrage Engine — cointegration, pairs trading, triangular arb, spread z-score
- Risk-Adjusted Compounding — Sharpe/Sortino/Calmar metrics, Kelly compounding, projections
- Performance Attribution — P&L decomposition by pair, setup, direction, session, day
- Monte Carlo Stress Testing — bootstrap, parametric (fat-tail), regime shuffle, parameter perturbation
- Parameter Sensitivity — 2D grid sweep, OAT analysis, heatmaps, overfit/flatness scoring
- Trade Psychology Coach — tilt detector, trade-psychology-coach bias check, discipline score, pre-trade checklist
- Trade-Level Risk Refinement — midpoint entry (SL halving), partial close & break-even protocol
Reference Files
- references/risk-and-portfolio.md — Core risk rules, position sizing (fixed %, Kelly, ATR), drawdown protocol, portfolio heat, psychology, journal template
- references/execution-costs.md — Entry timing, order splitting/scaling, trailing stops, market microstructure, spread/slippage analysis, market impact model
- references/journal-and-filter.md — Trade journal engine, performance analytics (win rate, equity curve, R-multiple), event blackout filter, should_i_trade()
- references/portfolio-allocation.md — Portfolio optimization (Markowitz, Black-Litterman, risk parity), strategy allocator, currency baskets, synthetic instruments
- references/account-tail-risk.md — Multi-account manager (aggregate P&L) + tail risk hedging (black swan protection)
- references/risk-premia.md — Factor investing: momentum factor, carry factor, value factor, volatility premium harvesting, combined factor portfolio
- references/arbitrage-engine.md — Cointegration testing, OLS hedge ratio, triangular arbitrage check, spread z-score signals, pair scanner
Quick Decision Guide
| Task | Load |
|---|---|
| Position sizing, lot size, stop placement | references/risk-and-portfolio.md |
| Kelly criterion, drawdown rules, psychology | references/risk-and-portfolio.md |
| When to enter, how to scale, slippage | references/execution-costs.md |
| Microstructure, spread analysis, TCA | references/execution-costs.md |
| Log a trade, check win rate, equity curve | references/journal-and-filter.md |
| Should I trade today? Event blackouts? | references/journal-and-filter.md |
| Portfolio weights, Markowitz, risk parity | references/portfolio-allocation.md |
| Currency basket, DXY replica, multi-strategy | references/portfolio-allocation.md |
| Multi-account P&L, broker comparison | references/account-tail-risk.md |
| Black swan hedge, crash protection | references/account-tail-risk.md |
| Factor investing, momentum/carry/value/vol premia | references/risk-premia.md |
| Cointegration, pairs trade, triangular arb, spread z-score | references/arbitrage-engine.md |
| Sharpe/Sortino/Calmar, compounding projections, Kelly growth | See Risk-Adjusted Compounding below |
| P&L by pair/setup/session/direction, edge analysis | See Performance Attribution Engine below |
| Monte Carlo bootstrap, parametric fat-tail, regime shuffle | See Monte Carlo Stress Tester below |
| Parameter grid sweep, heatmap, overfit detection | See Parameter Sensitivity Analyzer below |
| Tilt detection, revenge trading, trade-psychology-coach biases, discipline | See Trade Psychology Coach below |
| Midpoint entry (SL halving), partial close, break-even rule | See Trade-Level Risk Refinement below |
Core Quick Reference Card
POSITION SIZE = (Account × Risk%) / Stop Distance
KELLY % = W − [(1−W) / R] → Use HALF Kelly
ATR STOP = ATR × Multiplier (2–4× depending on TF)
EXPECTED VALUE = (Win% × Avg Win) − (Loss% × Avg Loss)
PROFIT FACTOR = Gross Profit / Gross Loss → Target > 1.5
DAILY LIMIT = 3–5% account loss → STOP
WEEKLY LIMIT = 5–10% → reduce size
MAX DRAWDOWN = 15–25% → halt and review
PORTFOLIO HEAT = 6% max total open risk
LOSING STREAK = 3 losses → cut 50% | 5 losses → 25% or pause
MIN R:R = 1:1.5 → prefer 1:2+
BREAKEVEN STOP = move to entry when +1R reached
KILL SWITCH = 10% daily DD → stop trading for the day
MARGIN = never exceed 50% utilization
Inline Implementations (merged from risk-and-portfolio)
The sections below contain full Python implementations for risk-adjusted compounding, tail risk hedging, spread/slippage cost analysis, performance attribution, Monte Carlo stress testing, parameter sensitivity analysis, trade psychology coaching, and trade-level risk refinement techniques.
Risk Adjusted Compounding
Risk Adjusted Compounding
import numpy as np
import pandas as pd
from typing import Optional
# ── Shared performance metrics ──────────────────────────────────────────────
def _sharpe(returns: np.ndarray, risk_free_daily: float = 0.04 / 252,
annualise: bool = True) -> float:
"""Annualised Sharpe ratio (excess return / vol)."""
excess = returns - risk_free_daily
std = returns.std(ddof=1)
if std == 0:
return 0.0
sr = excess.mean() / std
return float(sr * np.sqrt(252) if annualise else sr)
def _sortino(returns: np.ndarray, risk_free_daily: float = 0.04 / 252,
annualise: bool = True) -> float:
"""Sortino ratio — penalises only downside volatility."""
excess = returns - risk_free_daily
downside = returns[returns < 0]
down_std = downside.std(ddof=1) if len(downside) > 1 else 0.0
if down_std == 0:
return 0.0
sr = excess.mean() / down_std
return float(sr * np.sqrt(252) if annualise else sr)
def _calmar(returns: np.ndarray) -> float:
"""Calmar ratio = annualised return / max drawdown."""
equity = np.cumprod(1 + returns)
peak = np.maximum.accumulate(equity)
dd = (equity - peak) / peak
max_dd = abs(dd.min())
ann_ret = (equity[-1] ** (252 / len(returns)) - 1)
return float(ann_ret / max_dd) if max_dd > 0 else 0.0
def _max_drawdown(returns: np.ndarray) -> float:
equity = np.cumprod(1 + returns)
peak = np.maximum.accumulate(equity)
return float(((equity - peak) / peak).min())
def _profit_factor(returns: np.ndarray) -> float:
gains = returns[returns > 0].sum()
losses = abs(returns[returns < 0].sum())
return float(gains / losses) if losses > 0 else float("inf")
def comprehensive_metrics(
returns: "pd.Series | np.ndarray",
risk_free_annual: float = 0.04,
label: str = "Strategy",
) -> dict:
"""
Compute a full suite of risk-adjusted performance metrics.
Parameters
----------
returns : daily returns series (decimal, e.g. 0.01 = +1 %)
risk_free_annual : annual risk-free rate (default 4 %)
label : name shown in output
Returns
-------
dict with Sharpe, Sortino, Calmar, max drawdown, profit factor, VaR, CVaR, etc.
Example
-------
>>> metrics = comprehensive_metrics(df["returns"], label="My Strategy")
>>> print(metrics["sharpe"])
1.42
"""
r = np.asarray(returns, dtype=float)
r = r[~np.isnan(r)]
if len(r) < 5:
return {"error": "Need at least 5 return observations", "label": label}
rfr_daily = risk_free_annual / 252
equity = np.cumprod(1 + r)
ann_ret = float(equity[-1] ** (252 / len(r)) - 1)
# VaR & CVaR
var_95 = float(np.percentile(r, 5))
cvar_95 = float(r[r <= var_95].mean()) if (r <= var_95).any() else var_95
return {
"label": label,
"n_periods": len(r),
"total_return": round(float(equity[-1] - 1) * 100, 2),
"ann_return": round(ann_ret * 100, 2),
"ann_volatility": round(float(r.std(ddof=1) * np.sqrt(252)) * 100, 2),
"sharpe": round(_sharpe(r, rfr_daily), 3),
"sortino": round(_sortino(r, rfr_daily), 3),
"calmar": round(_calmar(r), 3),
"max_drawdown": round(_max_drawdown(r) * 100, 2),
"profit_factor": round(_profit_factor(r), 3),
"win_rate": round(float((r > 0).mean()) * 100, 1),
"var_95": round(var_95 * 100, 3),
"cvar_95": round(cvar_95 * 100, 3),
"skewness": round(float(pd.Series(r).skew()), 3),
"kurtosis": round(float(pd.Series(r).kurtosis()), 3),
"grade": (
"A+" if _sharpe(r, rfr_daily) > 2.0 else
"A" if _sharpe(r, rfr_daily) > 1.5 else
"B" if _sharpe(r, rfr_daily) > 1.0 else
"C" if _sharpe(r, rfr_daily) > 0.5 else "D"
),
}
class CompoundingOptimizer:
"""
Kelly-criterion compounding and multi-period projection.
Example
-------
>>> opt = CompoundingOptimizer.optimal_growth(0.55, 1.5, 1.0)
>>> print(opt["recommended"])
8.33
"""
@staticmethod
def optimal_growth(
win_rate: float,
avg_win: float,
avg_loss: float,
trades_per_year: int = 252,
) -> dict:
"""
Compute Kelly criterion fractions and geometric growth rates.
Parameters
----------
win_rate : fraction of winning trades (0–1)
avg_win : average win in R-multiples (or % of account)
avg_loss : average loss in R-multiples (positive value)
trades_per_year : used for annualising growth rate
Returns
-------
dict with full/half/quarter kelly and projected growth rates.
"""
if avg_win <= 0:
raise ValueError("avg_win must be positive")
if not 0 < win_rate < 1:
raise ValueError("win_rate must be between 0 and 1 (exclusive)")
kelly = (win_rate * avg_win - (1 - win_rate) * avg_loss) / avg_win
kelly = max(kelly, 0.0) # No negative sizing
half_kelly = kelly / 2
quarter_kelly = kelly / 4
def _growth_rate(f: float) -> float:
"""Expected log-growth per trade at fraction f."""
try:
return (win_rate * np.log1p(f * avg_win / 100)
+ (1 - win_rate) * np.log1p(-f * avg_loss / 100))
except (ValueError, ZeroDivisionError):
return -np.inf
return {
"full_kelly_pct": round(kelly * 100, 2),
"half_kelly_pct": round(half_kelly * 100, 2),
"quarter_kelly_pct": round(quarter_kelly * 100, 2),
"recommended": round(half_kelly * 100, 2),
"expectancy_r": round(win_rate * avg_win - (1 - win_rate) * avg_loss, 4),
"growth_at_half_kelly_ann": round(_growth_rate(half_kelly) * trades_per_year * 100, 2),
"growth_at_quarter_kelly_ann": round(_growth_rate(quarter_kelly) * trades_per_year * 100, 2),
"risk_of_ruin_half_kelly": round(
((1 - win_rate) / win_rate) ** (1 / max(half_kelly, 1e-6)), 4
),
"note": (
"Half Kelly = ~75 % of full Kelly growth with ~50 % of the variance. "
"Best risk-adjusted choice for most traders."
),
}
@staticmethod
def compound_projection(
balance: float,
monthly_return_pct: float,
months: int = 12,
withdrawal_pct: float = 0.0,
) -> dict:
"""
Project compounded account growth with optional monthly withdrawals.
Parameters
----------
balance : starting capital
monthly_return_pct: expected monthly return (%)
months : projection horizon
withdrawal_pct : % of balance withdrawn each month (0 = reinvest all)
Returns
-------
dict with ending balance, total return, CAGR, and monthly curve.
"""
if balance <= 0:
raise ValueError("balance must be positive")
curve = [balance]
r = monthly_return_pct / 100
w = withdrawal_pct / 100
for _ in range(months):
prev = curve[-1]
after = prev * (1 + r) * (1 - w)
curve.append(after)
ending = curve[-1]
cagr_ann = ((ending / balance) ** (12 / max(months, 1)) - 1) * 100
return {
"starting": balance,
"ending": round(ending, 2),
"total_return_pct": round((ending / balance - 1) * 100, 2),
"cagr_annual_pct": round(cagr_ann, 2),
"months": months,
"monthly_curve": [round(v, 2) for v in curve],
"withdrawal_total": round(sum(curve[i] * w for i in range(months)), 2),
}
Tail Risk Hedging
Tail Risk Hedging
import numpy as np, pandas as pd
class TailRiskHedging:
"""
Quantify tail risk and recommend hedging strategies.
Uses vectorised numpy for all statistical calculations.
Supports both empirical VaR/CVaR and parametric (Cornish-Fisher) estimates.
Example
-------
>>> risk = TailRiskHedging.assess_tail_risk(returns)
>>> print(risk["risk_level"], risk["cvar_99"])
'HIGH' -3.21
"""
@staticmethod
def assess_tail_risk(returns: "pd.Series | np.ndarray") -> dict:
"""
Compute full tail-risk profile from a return series.
Parameters
----------
returns : daily returns (decimal, e.g. -0.02 = -2 %)
Returns
-------
dict with kurtosis, skewness, VaR/CVaR at 95 % and 99 %,
worst daily/weekly, Cornish-Fisher adjusted VaR, and risk_level.
"""
r = np.asarray(returns, dtype=float)
r = r[~np.isnan(r)]
if len(r) < 10:
return {"error": "Need at least 10 observations"}
s = pd.Series(r)
kurt = float(s.kurtosis())
skew = float(s.skew())
mu = float(r.mean())
sig = float(r.std(ddof=1))
# Historical VaR / CVaR
var_95 = float(np.percentile(r, 5))
var_99 = float(np.percentile(r, 1))
cvar_95 = float(r[r <= var_95].mean()) if (r <= var_95).any() else var_95
cvar_99 = float(r[r <= var_99].mean()) if (r <= var_99).any() else var_99
# Cornish-Fisher adjusted VaR (accounts for skewness & kurtosis)
z_95 = -1.645
z_99 = -2.326
cf_adj = lambda z: (z + (z**2 - 1) * skew / 6 +
(z**3 - 3*z) * (kurt - 3) / 24 -
(2*z**3 - 5*z) * skew**2 / 36)
cf_var_95 = float(mu + cf_adj(z_95) * sig)
cf_var_99 = float(mu + cf_adj(z_99) * sig)
# Rolling-window worst metrics
worst_day = float(r.min())
worst_week = float(s.rolling(5).sum().min())
# Max drawdown on the return series
equity = np.cumprod(1 + r)
peak = np.maximum.accumulate(equity)
max_dd = float(((equity - peak) / peak).min())
risk_level = (
"EXTREME" if kurt > 10 or cvar_99 < -0.05 else
"HIGH" if kurt > 5 or cvar_99 < -0.03 else
"MODERATE"
)
return {
"n_observations": len(r),
"kurtosis": round(kurt, 3),
"skewness": round(skew, 3),
"fat_tails": kurt > 3,
"var_95": round(var_95 * 100, 3),
"var_99": round(var_99 * 100, 3),
"cvar_95": round(cvar_95 * 100, 3),
"cvar_99": round(cvar_99 * 100, 3),
"cf_var_95": round(cf_var_95 * 100, 3),
"cf_var_99": round(cf_var_99 * 100, 3),
"worst_daily_loss": round(worst_day * 100, 3),
"worst_weekly_loss": round(worst_week * 100, 3),
"max_drawdown_pct": round(max_dd * 100, 2),
"risk_level": risk_level,
}
@staticmethod
def hedging_strategies(portfolio_size: float, risk_budget_pct: float = 2.0,
vix_level: float = 20.0) -> dict:
"""
Recommend hedging strategies based on portfolio size, risk budget, and VIX.
Parameters
----------
portfolio_size : total account value in USD
risk_budget_pct : % of portfolio allocated to hedging
vix_level : current VIX for regime-based recommendations
Returns
-------
dict with hedge_budget, ranked strategies, trigger_rules, and urgency.
"""
if portfolio_size <= 0:
raise ValueError("portfolio_size must be positive")
hedge_budget = portfolio_size * risk_budget_pct / 100
urgency = (
"CRITICAL" if vix_level > 40 else
"HIGH" if vix_level > 30 else
"ELEVATED" if vix_level > 22 else
"NORMAL"
)
return {
"hedge_budget": round(hedge_budget, 2),
"vix_level": vix_level,
"urgency": urgency,
"strategies": [
{
"name": "Safe haven allocation",
"method": "Hold 5-10% in JPY, CHF, Gold positions",
"cost": "Low (may earn carry on some)",
"protection": "Moderate",
"recommended_at": "VIX > 20",
},
{
"name": "Correlated short hedge",
"method": "Short small position in highly correlated pair",
"cost": "Spread + potential adverse move",
"protection": "Moderate",
"recommended_at": "VIX > 25",
},
{
"name": "Reduce gross exposure",
"method": "Cut all positions to 50% during high-VIX",
"cost": "Opportunity cost",
"protection": "High",
"recommended_at": "VIX > 30",
},
{
"name": "Tail-triggered stop-all",
"method": "Auto-close everything if portfolio DD > 5% intraday",
"cost": "Slippage on emergency close",
"protection": "Maximum",
"recommended_at": "VIX > 35",
},
{
"name": "Diversification across timeframes",
"method": "Mix scalp + swing + position strategies",
"cost": "Management complexity",
"protection": "Moderate — decorrelates drawdowns",
"recommended_at": "Always",
}],
"trigger_rules": {
"VIX_above_30": "Reduce all positions to 50%",
"VIX_above_40": "Close all positions, go to cash",
"correlation_spike_above_0.9": "Close correlated positions — diversification broken",
"daily_DD_above_3pct": "Stop trading, review positions",
},
"active_recommendations": (
["Reduce gross exposure", "Tail-triggered stop-all"]
if vix_level > 30 else
["Correlated short hedge", "Reduce gross exposure"]
if vix_level > 22 else
["Safe haven allocation", "Diversification across timeframes"]
),
}
Spread Slippage Cost Analyzer
Spread & Slippage Cost Analyzer
import pandas as pd
import numpy as np
class CostAnalyzer:
"""
Analyse transaction costs: spreads, slippage, and their drag on strategy returns.
All computations are vectorised. Provides breakeven analysis so traders know
the minimum edge needed to overcome costs.
Example
-------
>>> stats = CostAnalyzer.spread_statistics(tick_df)
>>> print(stats["avg_spread_pips"], stats["p95_spread_pips"])
1.2 3.4
"""
@staticmethod
def spread_statistics(ticks: pd.DataFrame, pip_factor: float = 10_000.0) -> dict:
"""
Full spread distribution analysis from tick data.
Parameters
----------
ticks : DataFrame with 'ask' and 'bid' columns and DatetimeIndex
pip_factor : multiplier to convert price diff to pips (10000 for 5-digit pairs)
Returns
-------
dict with full spread distribution, widening events, and quality grade.
"""
if "ask" not in ticks.columns or "bid" not in ticks.columns:
raise ValueError("ticks DataFrame must have 'ask' and 'bid' columns")
spread_pips = (ticks["ask"] - ticks["bid"]) * pip_factor
avg = float(spread_pips.mean())
p95 = float(spread_pips.quantile(0.95))
grade = (
"EXCELLENT" if avg < 1.0 else
"GOOD" if avg < 2.0 else
"FAIR" if avg < 3.5 else
"POOR"
)
return {
"avg_spread_pips": round(avg, 3),
"median_spread_pips": round(float(spread_pips.median()), 3),
"min_spread_pips": round(float(spread_pips.min()), 3),
"max_spread_pips": round(float(spread_pips.max()), 3),
"std_spread_pips": round(float(spread_pips.std(ddof=1)), 3),
"p95_spread_pips": round(p95, 3),
"p99_spread_pips": round(float(spread_pips.quantile(0.99)), 3),
"spread_widening_events": int((spread_pips > p95).sum()),
"pct_time_above_2x_avg": round(float((spread_pips > 2 * avg).mean()) * 100, 1),
"spread_quality_grade": grade,
"n_ticks": len(ticks),
}
@staticmethod
def spread_by_session(ticks: pd.DataFrame, pip_factor: float = 10_000.0) -> dict:
"""
Spread behaviour per trading session — find the cheapest time to trade.
Returns
-------
dict of session → {mean, median, max} spread in pips.
"""
if not hasattr(ticks.index, "hour"):
raise ValueError("ticks must have a DatetimeIndex")
t = ticks.copy()
t["spread_pips"] = (t["ask"] - t["bid"]) * pip_factor
t["session"] = t.index.hour.map(
lambda h: (
"tokyo" if h < 7 else
"london" if h < 13 else
"overlap" if h < 16 else
"ny_late" if h < 22 else
"off"
)
)
stats = (
t.groupby("session")["spread_pips"]
.agg(mean="mean", median="median", max="max", p95=lambda x: x.quantile(0.95))
.round(3)
)
result = stats.to_dict("index")
# Flag cheapest session
if result:
cheapest = min(result, key=lambda s: result[s].get("mean", 99))
result["_cheapest_session"] = cheapest
return result
@staticmethod
def slippage_analysis(trades: pd.DataFrame, pip_factor: float = 10_000.0) -> dict:
"""
Analyse actual slippage from trade execution data.
Parameters
----------
trades : DataFrame with 'expected_price', 'actual_price', and optionally 'lot_size'
Returns
-------
dict with average/max slippage, adverse vs favourable split, and breakeven impact.
"""
if "expected_price" not in trades.columns or "actual_price" not in trades.columns:
return {"error": "Need 'expected_price' and 'actual_price' columns"}
if len(trades) == 0:
return {"error": "No trades to analyse"}
t = trades.copy()
raw_slip = (t["actual_price"] - t["expected_price"]) * pip_factor
t["slippage"] = raw_slip.abs()
adverse = float((raw_slip > 0).mean() * 100) # bought higher / sold lower
avg_slip = float(t["slippage"].mean())
max_slip = float(t["slippage"].max())
total_pip = float(t["slippage"].sum())
# Per-lot USD cost estimate (1 pip ≈ $10 for standard lot)
usd_per_lot = avg_slip * 10.0 if "lot_size" not in t.columns else \
float((t["slippage"] * t["lot_size"] * 10.0).mean())
return {
"avg_slippage_pips": round(avg_slip, 3),
"max_slippage_pips": round(max_slip, 3),
"p95_slippage_pips": round(float(t["slippage"].quantile(0.95)), 3),
"pct_adverse_slippage": round(adverse, 1),
"pct_favourable_slippage": round(100 - adverse, 1),
"total_slippage_cost_pips": round(total_pip, 1),
"avg_slippage_usd_per_lot": round(usd_per_lot, 2),
"n_trades": len(t),
}
@staticmethod
def cost_impact_on_strategy(
avg_spread: float,
avg_slippage: float,
trades_per_year: int,
avg_profit_per_trade: float,
) -> dict:
"""
Calculate what fraction of gross profit is consumed by transaction costs.
Parameters
----------
avg_spread : average round-trip spread cost in pips
avg_slippage : average round-trip slippage in pips
trades_per_year : number of trades per year
avg_profit_per_trade : gross average profit per trade in pips
Returns
-------
dict with cost ratios, breakeven trade minimum, and verdict.
"""
cost_rt = avg_spread + avg_slippage
annual_cost = cost_rt * trades_per_year
cost_pct = cost_rt / max(abs(avg_profit_per_trade), 0.01) * 100
breakeven_trades = int(np.ceil(annual_cost / max(abs(avg_profit_per_trade), 0.01)))
return {
"cost_per_trade_pips": round(cost_rt, 2),
"annual_cost_pips": round(annual_cost, 1),
"cost_as_pct_of_profit": round(cost_pct, 1),
"net_profit_ratio": round(1 - cost_pct / 100, 4),
"breakeven_trades_year": breakeven_trades,
"min_profit_to_survive": round(cost_rt * 1.5, 2), # Need ≥ 1.5× cost to be viable
"verdict": (
"ACCEPTABLE — costs are manageable" if cost_pct < 30 else
"HIGH — reduce trades or find tighter broker" if cost_pct < 60 else
"CRITICAL — costs are destroying the edge"
),
}
@staticmethod
def broker_comparison(broker_data: list[dict]) -> pd.DataFrame:
"""
Compare multiple brokers on cost metrics, ranked by effective total cost.
broker_data : list of dicts with keys: broker, avg_spread_pips, commission_pips,
overnight_swap_long, overnight_swap_short
"""
df = pd.DataFrame(broker_data)
if "avg_spread_pips" not in df.columns:
raise ValueError("Each broker entry must have 'avg_spread_pips'")
comm_col = "commission_pips" if "commission_pips" in df.columns else None
df["total_cost_pips"] = (
df["avg_spread_pips"] + (df[comm_col] if comm_col else 0)
)
return df.sort_values("total_cost_pips").reset_index(drop=True)
Performance Attribution Engine
Performance Attribution Engine
import pandas as pd, numpy as np
class PerformanceAttribution:
"""
Decompose P&L into contributions by pair, setup, direction, day, session, and
time-of-entry — to identify where the true edge lives.
Example
-------
>>> report = PerformanceAttribution.attribute(journal_df)
>>> print(report["best_pair"], report["best_setup"])
'EURUSD' 'ICT_FVG'
"""
@staticmethod
def attribute(trades: pd.DataFrame) -> dict:
"""
Full P&L attribution across all available dimensions.
Parameters
----------
trades : DataFrame with at minimum 'pnl_usd' column.
Optional: 'symbol', 'setup_type', 'direction', 'entry_time', 'lot_size'
Returns
-------
dict with attribution by pair, setup, direction, day-of-week, session,
hour-of-day, win-rate breakdowns, and actionable insight.
"""
if trades.empty:
return {"error": "No trades"}
closed = trades[trades["pnl_usd"].notna()].copy()
if closed.empty:
return {"error": "No closed trades with P&L data"}
result: dict = {}
# ── P&L attribution by dimension ──
def _attr(col: str) -> dict:
if col not in closed.columns:
return {}
grp = closed.groupby(col)
pnl = grp["pnl_usd"].sum().sort_values(ascending=False)
count = grp["pnl_usd"].count()
wr = grp["pnl_usd"].apply(lambda x: (x > 0).mean() * 100).round(1)
return {
k: {
"total_pnl": round(float(pnl[k]), 2),
"n_trades": int(count[k]),
"win_rate": float(wr[k]),
}
for k in pnl.index
}
result["by_pair"] = _attr("symbol")
result["by_setup"] = _attr("setup_type")
result["by_direction"] = _attr("direction")
# Day-of-week
if "entry_time" in closed.columns:
closed["_dow"] = pd.to_datetime(closed["entry_time"]).dt.day_name()
closed["_hour"] = pd.to_datetime(closed["entry_time"]).dt.hour
closed["_session"] = closed["_hour"].map(
lambda h: "tokyo" if h < 7 else "london" if h < 13 else "overlap" if h < 16 else "ny" if h < 22 else "off"
)
result["by_day"] = _attr("_dow")
result["by_hour"] = _attr("_hour")
result["by_session"] = _attr("_session")
# Lot-size consistency (risk management check)
if "lot_size" in closed.columns:
lots = closed["lot_size"]
result["lot_size_stats"] = {
"mean": round(float(lots.mean()), 2),
"std": round(float(lots.std(ddof=1)), 2),
"cv_pct": round(float(lots.std(ddof=1) / max(lots.mean(), 1e-10) * 100), 1),
"note": "CV > 50% suggests inconsistent sizing" if lots.std(ddof=1) / max(lots.mean(), 1e-10) > 0.5 else "Sizing is consistent",
}
# Summary leaders
bp = result["by_pair"]
bs = result["by_setup"]
result["best_pair"] = max(bp, key=lambda k: bp[k]["total_pnl"]) if bp else None
result["worst_pair"] = min(bp, key=lambda k: bp[k]["total_pnl"]) if bp else None
result["best_setup"] = max(bs, key=lambda k: bs[k]["total_pnl"]) if bs else None
result["worst_setup"] = min(bs, key=lambda k: bs[k]["total_pnl"]) if bs else None
result["overall_pnl"] = round(float(closed["pnl_usd"].sum()), 2)
result["overall_trades"] = len(closed)
result["overall_wr"] = round(float((closed["pnl_usd"] > 0).mean() * 100), 1)
result["insight"] = (
f"Best edge: {result['best_pair']} / {result['best_setup']}. "
f"Eliminate or reduce: {result['worst_pair']} / {result['worst_setup']}. "
"Concentrate capital on what works."
)
return result
Monte Carlo Stress Tester
Monte Carlo Stress Tester
import numpy as np
import pandas as pd
from typing import Callable, Optional
class MonteCarloStressTester:
"""
Vectorised Monte Carlo stress testing for trading strategies.
All simulations use numpy vectorisation — no Python loops over paths.
Example
-------
>>> result = MonteCarloStressTester.bootstrap_returns(returns, n_sims=5000)
>>> print(result["sharpe_median"])
1.23
"""
@staticmethod
def _batch_metrics(paths: np.ndarray, initial: float) -> dict:
"""
Compute metrics across all simulation paths — fully vectorised.
Parameters
----------
paths : (n_sims, n_steps) equity array
initial : starting equity value
Returns
-------
dict with median/percentile finals and drawdowns.
"""
finals = paths[:, -1] # (n_sims,)
peaks = np.maximum.accumulate(paths, axis=1) # (n_sims, n_steps)
dds = (paths - peaks) / peaks # (n_sims, n_steps)
max_dds = dds.min(axis=1) # (n_sims,)
# Annualised returns per path
n_steps = paths.shape[1]
ann_rets = (finals / initial) ** (252 / n_steps) - 1
# Per-path Sharpe (approximate — log returns on each path)
log_ret = np.diff(np.log(paths), axis=1) # (n_sims, n_steps-1)
path_sr = (log_ret.mean(axis=1) / (log_ret.std(axis=1, ddof=1) + 1e-10)) * np.sqrt(252)
return {
"median_final": round(float(np.median(finals)), 2),
"mean_final": round(float(finals.mean()), 2),
"p5_final": round(float(np.percentile(finals, 5)), 2),
"p95_final": round(float(np.percentile(finals, 95)), 2),
"prob_profit": round(float((finals > initial).mean() * 100), 1),
"prob_ruin_50": round(float((finals < initial * 0.5).mean() * 100), 2),
"prob_ruin_25": round(float((finals < initial * 0.25).mean() * 100), 2),
"median_max_dd": round(float(np.median(max_dds) * 100), 2),
"worst_max_dd": round(float(max_dds.min() * 100), 2),
"p95_max_dd": round(float(np.percentile(max_dds, 5) * 100), 2),
"sharpe_median": round(float(np.median(path_sr)), 3),
"sharpe_p5": round(float(np.percentile(path_sr, 5)), 3),
"ann_return_p50": round(float(np.median(ann_rets) * 100), 2),
}
@staticmethod
def bootstrap_returns(
returns: pd.Series,
n_sims: int = 5000,
n_days: int = 252,
initial: float = 10_000,
seed: int = 42,
) -> dict:
"""
Bootstrap resampling — randomly shuffle return order to test path dependency.
Uses fully vectorised numpy sampling (no Python for-loop over paths).
Parameters
----------
returns : daily return Series
n_sims : number of Monte Carlo paths
n_days : path length in trading days
initial : starting equity
seed : random seed for reproducibility
Returns
-------
dict with distributional statistics across all paths.
"""
rng = np.random.default_rng(seed)
r_arr = returns.dropna().values
# Vectorised sample: draw entire (n_sims × n_days) matrix at once
sampled = rng.choice(r_arr, size=(n_sims, n_days), replace=True)
paths = initial * np.cumprod(1 + sampled, axis=1)
result = MonteCarloStressTester._batch_metrics(paths, initial)
result["type"] = "bootstrap"
result["n_sims"] = n_sims
return result
@staticmethod
def parametric_simulation(
returns: pd.Series,
n_sims: int = 5_000,
n_days: int = 252,
initial: float = 10_000,
fat_tails: bool = True,
seed: int = 42,
) -> dict:
"""
Parametric Monte Carlo using fitted normal (or Student-t for fat tails).
Parameters
----------
fat_tails : if True, uses Student-t distribution (df=5) instead of normal
Returns
-------
dict with same keys as bootstrap_returns, plus distribution parameters.
"""
rng = np.random.default_rng(seed)
r_clean = returns.dropna().values
mu = r_clean.mean()
sigma = r_clean.std(ddof=1)
if fat_tails:
from scipy.stats import t as t_dist
df_fit = 5.0 # Student-t degrees of freedom (fat tails)
# Scale t-distributed samples to match empirical mu/sigma
raw = rng.standard_t(df=df_fit, size=(n_sims, n_days))
samples = mu + sigma * raw / np.sqrt(df_fit / (df_fit - 2))
else:
samples = rng.normal(mu, sigma, size=(n_sims, n_days))
paths = initial * np.cumprod(1 + samples, axis=1)
result = MonteCarloStressTester._batch_metrics(paths, initial)
result.update({
"type": "parametric_fat_tails" if fat_tails else "parametric_normal",
"n_sims": n_sims,
"fitted_mu": round(mu, 6),
"fitted_sigma": round(sigma, 6),
})
return result
@staticmethod
def parameter_perturbation(
strategy_fn: Callable,
base_params: dict,
data: pd.DataFrame,
perturbation_pct: float = 0.1,
n_tests: int = 100,
seed: int = 42,
) -> dict:
"""
Randomly perturb strategy parameters ±X% and check robustness.
Parameters
----------
strategy_fn : callable(data, params) → pd.Series of returns
base_params : baseline parameter dict
data : OHLCV DataFrame
perturbation_pct: fraction of each param to perturb (0.1 = ±10 %)
n_tests : number of random perturbations
Returns
-------
dict with Sharpe distribution and robustness verdict.
"""
rng = np.random.default_rng(seed)
sharpes: list[float] = []
for _ in range(n_tests):
perturbed = {
k: (type(v)(v * (1 + perturbation_pct * rng.uniform(-1, 1)))
if isinstance(v, (int, float)) else v)
for k, v in base_params.items()
}
try:
ret = strategy_fn(data, perturbed)
r = np.asarray(ret, dtype=float)
r = r[~np.isnan(r)]
if len(r) < 5:
continue
sharpe = float((r.mean() / r.std(ddof=1)) * np.sqrt(252)) if r.std(ddof=1) > 0 else 0.0
sharpes.append(sharpe)
except Exception:
pass # Silently skip invalid parameter combinations
if not sharpes:
return {"type": "parameter_perturbation", "error": "No valid perturbations"}
s_arr = np.array(sharpes)
pct_pos = float((s_arr > 0).mean() * 100)
return {
"type": "parameter_perturbation",
"perturbation_pct": perturbation
…(truncated)