backtester/portfolio/crisis_analysis.py crisis_analysis.py:
Historical stress-period analysis for strategy and benchmark behavior.
This file loads crisis period definitions from JSON and computes per-event strategy behavior: return, volatility, drawdown, Sharpe, best/worst day, benchmark comparison and beta during crisis.
from backtester.portfolio.crisis_analysis import compute_crisis_analysis When To Read This
- 01You need to explain how the crisis table in a PDF report is produced.
- 02You want to add or override stress periods without changing source code.
- 03You are reviewing whether a strategy survives named market events.
File Anatomy
- CrisisPeriod dataclass: name, start, end and category.
- CrisisPeriodResult dataclass: strategy, benchmark and relative analytics for one event.
- Config loader: reads bundled or custom crisis_periods.json.
- Main entry point: loops through events, analyzes overlap and returns summary plus details.
Data Contract
Inputs
- returns: daily strategy returns Series.
- benchmark_returns: optional daily benchmark returns Series.
- config_path: optional custom JSON path.
- include_regimes: whether to include broader market regimes in addition to crisis periods.
Outputs
- summary: list of table-ready rows.
- details: OrderedDict period name -> full result dictionary.
- aggregate fields: overlapping_count, total_defined, average/worst/best crisis and hit rate.
Invariants
- Periods with fewer than two overlapping strategy observations are skipped.
- Timezone is normalized to the return index before slicing.
- Benchmark fields are optional and only populated when benchmark overlap exists.
Public API And Key Internals
CrisisPeriod
dataclassCrisisPeriod(name, start, end, category="market_crash") Historical event definition loaded from config.
CrisisPeriodResult
dataclassCrisisPeriodResult(name, category, start, end, trading_days, strategy_return, strategy_vol, strategy_max_dd, strategy_sharpe, strategy_worst_day, strategy_best_day, benchmark_return=None, ...) Analytics record for one event, with to_dict() for report serialization.
load_crisis_periods
functionload_crisis_periods(config_path=None, include_regimes=False) Loads bundled or custom JSON event definitions.
Returns
list[CrisisPeriod].
compute_crisis_analysis
functioncompute_crisis_analysis(returns, benchmark_returns=None, config_path=None, include_regimes=False) Runs the complete event analysis and returns table rows plus aggregate summary.
Returns
dict[str, Any].
_analyze_period
helper_analyze_period(period, strategy_returns, benchmark_returns) Internal function that computes one event window.
Returns
CrisisPeriodResult or None.
Implementation Notes
- The JSON config makes crisis definitions reviewable and replaceable without changing code.
- The result object includes best and worst single day, which helps explain whether damage was gradual or one-session.
- The aggregate summary is designed to be displayed in both rich report tables and PDF metrics.
Code Walkthrough
Run default crisis analysis
This returns the exact shape used by report tables and crisis plots.
from backtester.portfolio.crisis_analysis import compute_crisis_analysis
crisis_packet = compute_crisis_analysis(
returns=strategy_returns,
benchmark_returns=benchmark_returns,
include_regimes=True,
)
table_rows = crisis_packet["summary"]
worst_event = crisis_packet["worst_crisis"]
hit_rate = crisis_packet["crisis_hit_rate"] Key implementation: per-period result fields
The result captures path risk and benchmark-relative behavior inside each named event.
result = CrisisPeriodResult(
name=period.name,
category=period.category,
start=period.start.strftime("%Y-%m-%d"),
end=period.end.strftime("%Y-%m-%d"),
trading_days=len(strat),
strategy_return=float((1 + strat).prod() - 1),
strategy_vol=float(strat.std() * np.sqrt(252)),
strategy_max_dd=_max_drawdown(strat),
strategy_sharpe=_annualize_sharpe(strat),
strategy_worst_day=float(strat.min()),
strategy_best_day=float(strat.max()),
) Key implementation: report-ready return shape
The output is not just a DataFrame; it includes table rows, full details and aggregate review numbers.
return {
"summary": summary_rows,
"details": details,
"overlapping_count": len(results),
"total_defined": len(periods),
"avg_crisis_return": float(np.mean(strat_returns)) if strat_returns else 0.0,
"worst_crisis": results[int(np.argmin(strat_returns))].name if strat_returns else "N/A",
"best_crisis": results[int(np.argmax(strat_returns))].name if strat_returns else "N/A",
"crisis_hit_rate": len(positive) / len(strat_returns) * 100 if strat_returns else 0.0,
}