Gahow Wang
541f7bcf5b
Add 28 research scripts covering DCA simulation, momentum evaluation, Sharpe optimization, trend rider analysis, and US fundamentals exploration.
67 lines
2.2 KiB
Python
67 lines
2.2 KiB
Python
"""Trace where V3/V5 maximum drawdowns occur and what holdings they had."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
from itertools import product
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from research.trend_rider_robustness import (
|
|
load_price_panel,
|
|
portfolio_returns,
|
|
)
|
|
from strategies.permanent import TrendRiderV3
|
|
from strategies.trend_rider_v5 import TrendRiderV5
|
|
|
|
|
|
def trace(name: str, weights: pd.DataFrame, prices: pd.DataFrame,
|
|
start: str = "2015-01-02") -> None:
|
|
rets = portfolio_returns(weights, prices[weights.columns], 0.001)
|
|
rets = rets[rets.index >= start]
|
|
eq = (1 + rets).cumprod()
|
|
dd = eq / eq.cummax() - 1
|
|
trough = dd.idxmin()
|
|
peak = eq.loc[:trough].idxmax()
|
|
recover = eq.loc[trough:][eq.loc[trough:] >= eq.loc[peak]]
|
|
rec_dt = recover.index[0] if len(recover) else None
|
|
|
|
print(f"\n=== {name} ===")
|
|
print(f" MDD = {dd.min()*100:.2f}%")
|
|
print(f" Peak : {peak.date()} equity={eq.loc[peak]:.3f}")
|
|
print(f" Trough: {trough.date()} equity={eq.loc[trough]:.3f}")
|
|
print(f" Recovered: {rec_dt.date() if rec_dt is not None else 'NOT YET'}")
|
|
print(f" Days to trough: {(trough - peak).days}")
|
|
|
|
# Show holdings around the drawdown
|
|
print(f"\n Holdings 5 days before peak through 5 days after trough:")
|
|
sl = weights.loc[peak - pd.Timedelta(days=10): trough + pd.Timedelta(days=10)]
|
|
nonzero = (sl != 0).any(axis=0)
|
|
sl = sl.loc[:, nonzero]
|
|
sl_disp = sl.copy()
|
|
# Show only days when holdings change
|
|
changes = (sl_disp != sl_disp.shift(1)).any(axis=1)
|
|
sl_disp = sl_disp.loc[changes]
|
|
print(sl_disp.round(3).head(40).to_string())
|
|
|
|
|
|
def main() -> None:
|
|
prices = load_price_panel()
|
|
print(f"Panel: {prices.index.min().date()} to {prices.index.max().date()}")
|
|
|
|
candidates = {
|
|
"V3 default": TrendRiderV3(),
|
|
"V5 default (panic 1.6/4%)": TrendRiderV5(),
|
|
"V5 panic 1.8/5%": TrendRiderV5(panic_vol_ratio=1.8, panic_peak_drop_pct=0.05),
|
|
}
|
|
for name, strat in candidates.items():
|
|
w = strat.generate_signals(prices)
|
|
trace(name, w, prices)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|