# Risk And Portfolio > Complete trading risk management, portfolio construction, performance tracking, and quantitative stress testing — all in one skill. The safety layer for all trading decisions. RISK MANAGEMENT: "risk management", "position sizing", "lot size calculation", "stop loss placement", "drawdown management", "Kelly criterion", "ATR-based sizing", "fixed percentage risk model", "risk-reward analysis", "profit factor", "expectancy", "trading psychology", "trade-psychology-coach biases", "loss aversion", "FOMO", "revenge trading", "emotional management", "daily loss limit", "weekly loss limit", "max drawdown", "portfolio heat", "correlation risk", "trailing stops", "losing streak", "drawdown recovery", "risk of ruin", "VaR", "CVaR", "Calmar ratio", "Sortino ratio", "margin requirements", "money management", "1% rule trading". EXECUTION & COSTS: "best entry timing", "scaling in out", "limit vs market orders", "partial fills", "slippage reduction", "scale into position", "split my entry", "reduce slippage", "DCA strategy", - Skill: `mahmoud20138/risk-and-portfolio` (Agent Skill) - Install (CLI): `npx skillmds@latest add mahmoud20138/risk-and-portfolio` - Raw SKILL.md: https://api.skillmd.com/api/skills/mahmoud20138/risk-and-portfolio/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: AI & ML - Author: mahmoud20138 (https://skillmd.com/u/mahmoud20138) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/mahmoud20138/risk-and-portfolio --- > **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 1. **Risk Management** — position sizing, rules, psychology, drawdown recovery 2. **Execution & Costs** — entry timing, scaling, microstructure, spread/slippage 3. **Trade Journal & Performance** — logging, analytics, equity curve 4. **Trade Filter** — when NOT to trade (event blackouts, structural risks) 5. **Portfolio & Allocation** — Markowitz, risk parity, baskets, strategy allocation 6. **Multi-Account Manager** — aggregate P&L across brokers 7. **Tail Risk Hedging** — black swan protection 8. **Risk Premia** — factor investing (momentum, carry, value, volatility premium) 9. **Arbitrage Engine** — cointegration, pairs trading, triangular arb, spread z-score 10. **Risk-Adjusted Compounding** — Sharpe/Sortino/Calmar metrics, Kelly compounding, projections 11. **Performance Attribution** — P&L decomposition by pair, setup, direction, session, day 12. **Monte Carlo Stress Testing** — bootstrap, parametric (fat-tail), regime shuffle, parameter perturbation 13. **Parameter Sensitivity** — 2D grid sweep, OAT analysis, heatmaps, overfit/flatness scoring 14. **Trade Psychology Coach** — tilt detector, trade-psychology-coach bias check, discipline score, pre-trade checklist 15. **Trade-Level Risk Refinement** — midpoint entry (SL halving), partial close & break-even protocol --- ## Reference Files - **[references/risk-and-portfolio.md](references/risk-and-portfolio.md)** — Core risk rules, position sizing (fixed %, Kelly, ATR), drawdown protocol, portfolio heat, psychology, journal template - **[references/execution-costs.md](references/execution-costs.md)** — Entry timing, order splitting/scaling, trailing stops, market microstructure, spread/slippage analysis, market impact model - **[references/journal-and-filter.md](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](references/portfolio-allocation.md)** — Portfolio optimization (Markowitz, Black-Litterman, risk parity), strategy allocator, currency baskets, synthetic instruments - **[references/account-tail-risk.md](references/account-tail-risk.md)** — Multi-account manager (aggregate P&L) + tail risk hedging (black swan protection) - **[references/risk-premia.md](references/risk-premia.md)** — Factor investing: momentum factor, carry factor, value factor, volatility premium harvesting, combined factor portfolio - **[references/arbitrage-engine.md](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 ```python 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 ```python 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 ```python 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 ```python 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 ```python 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)