496 lines
22 KiB
Python
496 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from dataclasses import dataclass, asdict
|
|
from pathlib import Path
|
|
from typing import Dict, Tuple
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
ROOT = Path('/mnt/data')
|
|
OUT = ROOT / 'kartsell_v12_results'
|
|
OUT.mkdir(exist_ok=True)
|
|
|
|
FILES = {
|
|
'SPY': ROOT / 'SPY_Historical_Data.csv',
|
|
'AAPL': ROOT / 'AAPL.csv',
|
|
'AMZN': ROOT / 'AMZN.csv',
|
|
'NVDA': ROOT / 'NVDA.csv',
|
|
'TSLA': ROOT / 'TSLA.csv',
|
|
'AIG': ROOT / 'AIG.csv',
|
|
'AAMRQ': ROOT / 'AAMRQ.csv',
|
|
}
|
|
ASSET_CLASS = {
|
|
'SPY': 'broad_index_etf',
|
|
'AAPL': 'large_quality_stock',
|
|
'AMZN': 'large_quality_stock',
|
|
'NVDA': 'high_vol_stock',
|
|
'TSLA': 'high_vol_stock',
|
|
'AIG': 'cyclical_stock',
|
|
'AAMRQ': 'distressed_stock',
|
|
}
|
|
START = pd.Timestamp('2007-01-01')
|
|
END = pd.Timestamp('2026-08-01')
|
|
ONE_WAY_COST = 0.0007 # 7 bps: 2 bps commission + 5 bps slippage; FX/tax excluded.
|
|
|
|
|
|
def load_symbol(symbol: str, path: Path) -> pd.DataFrame:
|
|
df = pd.read_csv(path)
|
|
# Drop unnamed index columns.
|
|
df = df.loc[:, ~df.columns.astype(str).str.startswith('Unnamed')]
|
|
if df.columns[0] == '':
|
|
df = df.iloc[:, 1:]
|
|
rename = {c: c.strip().lower().replace('datetime', 'date') for c in df.columns}
|
|
df = df.rename(columns=rename)
|
|
if 'date' not in df.columns and 'DateTime' in df.columns:
|
|
df = df.rename(columns={'DateTime': 'date'})
|
|
df['date'] = pd.to_datetime(df['date'], errors='coerce', format='mixed')
|
|
for c in ['open', 'high', 'low', 'close', 'volume']:
|
|
df[c] = pd.to_numeric(df[c], errors='coerce')
|
|
df = df.dropna(subset=['date', 'open', 'high', 'low', 'close']).copy()
|
|
df = df[(df['date'] >= START) & (df['date'] <= END)]
|
|
df = df.sort_values('date').drop_duplicates('date', keep='last').reset_index(drop=True)
|
|
df['symbol'] = symbol
|
|
df['adj_close'] = df['close']
|
|
return df[['date','symbol','open','high','low','close','adj_close','volume']]
|
|
|
|
|
|
def data_quality(symbol: str, df: pd.DataFrame) -> dict:
|
|
raw = pd.read_csv(FILES[symbol])
|
|
date_col = 'DateTime' if 'DateTime' in raw.columns else 'date'
|
|
raw_dates = pd.to_datetime(raw[date_col], errors='coerce', format='mixed')
|
|
sorted_raw = raw_dates.dropna().is_monotonic_increasing
|
|
dup = raw_dates.duplicated().sum()
|
|
invalid_ohlc = ((df['high'] < df[['open','close','low']].max(axis=1)) |
|
|
(df['low'] > df[['open','close','high']].min(axis=1))).sum()
|
|
ret = df['close'].pct_change()
|
|
jumps50 = (ret.abs() > 0.50).sum()
|
|
gaps = df['date'].diff().dt.days
|
|
max_gap = int(gaps.max()) if len(gaps.dropna()) else 0
|
|
zero_vol = int((df['volume'].fillna(0) <= 0).sum())
|
|
return {
|
|
'symbol': symbol,
|
|
'start': df['date'].min().date().isoformat() if len(df) else None,
|
|
'end': df['date'].max().date().isoformat() if len(df) else None,
|
|
'rows': len(df),
|
|
'raw_sorted': bool(sorted_raw),
|
|
'raw_duplicate_dates': int(dup),
|
|
'invalid_ohlc_rows': int(invalid_ohlc),
|
|
'abs_daily_return_gt_50pct': int(jumps50),
|
|
'max_calendar_gap_days': max_gap,
|
|
'zero_or_missing_volume_rows': zero_vol,
|
|
'total_return_available': False,
|
|
'pit_fundamentals_available': False,
|
|
'corporate_action_table_available': False,
|
|
}
|
|
|
|
|
|
def add_features(df: pd.DataFrame) -> pd.DataFrame:
|
|
x = df.copy()
|
|
prev = x['close'].shift(1)
|
|
tr = pd.concat([(x['high']-x['low']), (x['high']-prev).abs(), (x['low']-prev).abs()], axis=1).max(axis=1)
|
|
x['atr20'] = tr.rolling(20, min_periods=20).mean()
|
|
x['sma50'] = x['close'].rolling(50, min_periods=50).mean()
|
|
x['sma200'] = x['close'].rolling(200, min_periods=200).mean()
|
|
x['sma50_slope20'] = x['sma50'] / x['sma50'].shift(20) - 1
|
|
x['ret1'] = x['close'].pct_change()
|
|
x['vol63'] = x['ret1'].rolling(63, min_periods=30).std() * math.sqrt(252)
|
|
x['vol252_med'] = x['vol63'].rolling(252, min_periods=63).median()
|
|
x['high20_prev'] = x['close'].rolling(20, min_periods=20).max().shift(1)
|
|
x['high63_prev'] = x['close'].rolling(63, min_periods=40).max().shift(1)
|
|
x['low20_prev'] = x['close'].rolling(20, min_periods=20).min().shift(1)
|
|
x['weekly_close'] = x.set_index('date')['close'].resample('W-FRI').last().reindex(x['date'], method='ffill').to_numpy()
|
|
return x
|
|
|
|
|
|
def activation_gain(asset_class: str) -> float:
|
|
return {
|
|
'broad_index_etf': 0.12,
|
|
'large_quality_stock': 0.18,
|
|
'high_vol_stock': 0.25,
|
|
'cyclical_stock': 0.22,
|
|
'distressed_stock': 0.30,
|
|
}[asset_class]
|
|
|
|
|
|
def core_floor(asset_class: str) -> float:
|
|
return {
|
|
'broad_index_etf': 0.55,
|
|
'large_quality_stock': 0.40,
|
|
'high_vol_stock': 0.25,
|
|
'cyclical_stock': 0.30,
|
|
'distressed_stock': 0.15,
|
|
}[asset_class]
|
|
|
|
|
|
def giveback_cap_v11(asset_class: str, peak_gain: float) -> float:
|
|
if asset_class == 'broad_index_etf':
|
|
return 0.12 if peak_gain < .25 else 0.10 if peak_gain < .50 else 0.08 if peak_gain < 1 else 0.07
|
|
if asset_class == 'large_quality_stock':
|
|
return 0.15 if peak_gain < .25 else 0.14 if peak_gain < .50 else 0.12 if peak_gain < 1 else 0.10
|
|
return 0.18 if peak_gain < .50 else 0.15 if peak_gain < 1 else 0.12
|
|
|
|
|
|
def giveback_cap_v12(asset_class: str, peak_gain: float) -> float:
|
|
# Deliberately wider than v11 to preserve secular winners; the portfolio floor handles systemic risk.
|
|
if asset_class == 'broad_index_etf':
|
|
return 0.14 if peak_gain < .25 else 0.12 if peak_gain < .50 else 0.10 if peak_gain < 1 else 0.08
|
|
if asset_class == 'large_quality_stock':
|
|
return 0.19 if peak_gain < .25 else 0.17 if peak_gain < .50 else 0.14 if peak_gain < 1 else 0.11
|
|
if asset_class == 'cyclical_stock':
|
|
return 0.20 if peak_gain < .50 else 0.17 if peak_gain < 1 else 0.14
|
|
if asset_class == 'distressed_stock':
|
|
return 0.28 if peak_gain < .50 else 0.23 if peak_gain < 1 else 0.18
|
|
return 0.25 if peak_gain < .50 else 0.20 if peak_gain < 1 else 0.16
|
|
|
|
|
|
def v11_positions(df: pd.DataFrame, asset_class: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
|
|
x = add_features(df)
|
|
pos = 1.0
|
|
entry = float(x.iloc[0]['close'])
|
|
peak = entry
|
|
prev_floor = -np.inf
|
|
breach_count = 0
|
|
last_sell_idx = -10000
|
|
rows, events = [], []
|
|
for i, row in x.iterrows():
|
|
px = float(row['close'])
|
|
peak = max(peak, px)
|
|
peak_gain = peak / entry - 1
|
|
atr = float(row['atr20']) if pd.notna(row['atr20']) else np.nan
|
|
dd = px / peak - 1
|
|
trend_bad = pd.notna(row['sma200']) and px < row['sma200']
|
|
regime = min(1, max(0, (-dd - .08)/.22))
|
|
flow = min(1, max(0, (float(row['vol63'])-.20)/.50)) if pd.notna(row['vol63']) else 0
|
|
lock = np.clip(.30 + .25*regime + .15*flow, .30, .85)
|
|
floor = np.nan
|
|
active = peak_gain >= activation_gain(asset_class)
|
|
if pd.notna(atr) and peak-entry >= 4*atr:
|
|
active = True
|
|
old = pos
|
|
reason = None
|
|
if active and pd.notna(atr):
|
|
atr_mult = np.clip(3.5-1.7*regime-.8*flow,1.6,3.5)
|
|
floor = max(entry + lock*(peak-entry), peak-atr_mult*atr, peak*(1-giveback_cap_v11(asset_class,peak_gain)), prev_floor)
|
|
prev_floor = floor
|
|
breach_count = breach_count+1 if px<floor else 0
|
|
gap=(floor-px)/atr if atr>0 else 0
|
|
if i-last_sell_idx>=5:
|
|
if gap>=1.5:
|
|
pos=max(.35,pos-.40); last_sell_idx=i; breach_count=0; reason='gap_floor'
|
|
elif breach_count>=2:
|
|
pos=max(.50,pos-.20); last_sell_idx=i; breach_count=0; reason='two_close_floor'
|
|
if trend_bad and dd<=-.18 and i-last_sell_idx>=5:
|
|
f=.35 if asset_class=='broad_index_etf' else .20
|
|
pos=max(f,pos-.25); last_sell_idx=i; reason='trend_crisis'
|
|
no_new_low=i>=5 and px>float(x.iloc[max(0,i-5):i+1]['close'].min())
|
|
recovery=pd.notna(row['sma200']) and px>row['sma200']
|
|
if pos<1 and no_new_low and recovery:
|
|
pos=min(1,pos+.15)
|
|
if pos>old: reason='daily_reentry'
|
|
if pos != old:
|
|
events.append({'date':row['date'],'old_position':old,'new_position':pos,'side':'BUY' if pos>old else 'SELL','reason':reason,'price':px,'peak':peak,'cycle_entry':entry,'floor':floor})
|
|
rows.append({'date':row['date'],'position':pos,'floor':floor,'entry':entry,'peak':peak})
|
|
return pd.DataFrame(rows), pd.DataFrame(events)
|
|
|
|
|
|
def v12_positions(df: pd.DataFrame, asset_class: str) -> Tuple[pd.DataFrame, pd.DataFrame]:
|
|
"""K-ArtSell 12.1 balanced candidate.
|
|
|
|
Keeps v11's strong profit floor, but fixes the stale-cycle and daily-refill defects:
|
|
- separated core floor by asset class;
|
|
- re-entry only after 10 sessions, 50D trend recovery and 20D breakout;
|
|
- no more than one 20%-point refill per 10 sessions;
|
|
- every refill starts a new tactical protection cycle.
|
|
"""
|
|
x = add_features(df)
|
|
pos = 1.0
|
|
entry = float(x.iloc[0]['close'])
|
|
peak = entry
|
|
prev_floor = -np.inf
|
|
breach_count = 0
|
|
last_sell = -10000
|
|
last_buy = -10000
|
|
below200_count = 0
|
|
rows, events = [], []
|
|
min_pos = {
|
|
'broad_index_etf': .35,
|
|
'large_quality_stock': .25,
|
|
'high_vol_stock': .20,
|
|
'cyclical_stock': .20,
|
|
'distressed_stock': .10,
|
|
}[asset_class]
|
|
crisis_dd = {
|
|
'broad_index_etf': -.18,
|
|
'large_quality_stock': -.18,
|
|
'high_vol_stock': -.22,
|
|
'cyclical_stock': -.20,
|
|
'distressed_stock': -.30,
|
|
}[asset_class]
|
|
|
|
for i, row in x.iterrows():
|
|
px = float(row['close'])
|
|
peak = max(peak, px)
|
|
peak_gain = peak / entry - 1.0
|
|
atr = float(row['atr20']) if pd.notna(row['atr20']) else np.nan
|
|
dd = px / peak - 1.0
|
|
below200_count = below200_count + 1 if pd.notna(row['sma200']) and px < row['sma200'] else 0
|
|
regime = np.clip((-dd - .08) / .22, 0, 1)
|
|
flow_proxy = np.clip((float(row['vol63']) - .20) / .50, 0, 1) if pd.notna(row['vol63']) else 0
|
|
floor = np.nan
|
|
old = pos
|
|
reason = None
|
|
|
|
active = peak_gain >= activation_gain(asset_class)
|
|
if pd.notna(atr) and peak - entry >= 4.0 * atr:
|
|
active = True
|
|
|
|
if active and pd.notna(atr):
|
|
lock = np.clip(.30 + .25 * regime + .15 * flow_proxy, .30, .85)
|
|
atr_mult = np.clip(3.5 - 1.7 * regime - .8 * flow_proxy, 1.6, 3.5)
|
|
floor = max(
|
|
entry + lock * (peak - entry),
|
|
peak - atr_mult * atr,
|
|
peak * (1.0 - giveback_cap_v11(asset_class, peak_gain)),
|
|
prev_floor,
|
|
)
|
|
prev_floor = floor
|
|
breach_count = breach_count + 1 if px < floor else 0
|
|
gap = (floor - px) / atr if atr > 0 else 0
|
|
if i - last_sell >= 5 and pos > min_pos:
|
|
if gap >= 1.5:
|
|
pos = max(min_pos, pos - .40)
|
|
last_sell = i
|
|
breach_count = 0
|
|
reason = 'gap_floor'
|
|
elif breach_count >= 2:
|
|
pos = max(min_pos, pos - .20)
|
|
last_sell = i
|
|
breach_count = 0
|
|
reason = 'two_close_floor'
|
|
|
|
if below200_count >= 1 and dd <= crisis_dd and i - last_sell >= 5 and pos > min_pos:
|
|
pos = max(min_pos, pos - .25)
|
|
last_sell = i
|
|
reason = 'trend_crisis'
|
|
|
|
# Staged, non-daily re-entry. Both trend recovery and a fresh breakout are required.
|
|
if pos < 1.0 and i - last_sell >= 10 and i - last_buy >= 10:
|
|
reclaim50 = (
|
|
pd.notna(row['sma50']) and px > row['sma50'] and
|
|
pd.notna(row['sma50_slope20']) and row['sma50_slope20'] > 0
|
|
)
|
|
breakout20 = pd.notna(row['high20_prev']) and px > row['high20_prev']
|
|
if reclaim50 and breakout20:
|
|
pos = min(1.0, pos + .20)
|
|
last_buy = i
|
|
reason = 'reentry_reclaim50_breakout20'
|
|
# New tactical capital must not inherit an obsolete historical floor.
|
|
entry = px
|
|
peak = px
|
|
prev_floor = -np.inf
|
|
breach_count = 0
|
|
|
|
if pos != old:
|
|
events.append({
|
|
'date': row['date'], 'old_position': old, 'new_position': pos,
|
|
'side': 'BUY' if pos > old else 'SELL', 'reason': reason,
|
|
'price': px, 'peak': peak, 'cycle_entry': entry, 'floor': floor,
|
|
})
|
|
rows.append({
|
|
'date': row['date'], 'position': pos, 'floor': floor,
|
|
'entry': entry, 'peak': peak, 'regime_proxy': regime,
|
|
'volshock_proxy': flow_proxy,
|
|
})
|
|
return pd.DataFrame(rows), pd.DataFrame(events)
|
|
|
|
def buyhold_positions(df):
|
|
return pd.DataFrame({'date':df['date'],'position':1.0})
|
|
|
|
|
|
def trend10m_positions(df):
|
|
x=df[['date','close']].copy()
|
|
m=x.set_index('date')['close'].resample('ME').last().dropna().to_frame('close')
|
|
m['sma10']=m['close'].rolling(10,min_periods=10).mean()
|
|
# Execute only after the month-end observation becomes known.
|
|
m['position']=(m['close']>=m['sma10']).astype(float).shift(1)
|
|
daily=m['position'].reindex(x['date'],method='ffill').fillna(1.0).to_numpy()
|
|
return pd.DataFrame({'date':x['date'],'position':daily})
|
|
|
|
|
|
def trailing_positions(df, trail=.12):
|
|
pos=1.0; peak=float(df.iloc[0]['close']); cooldown=0; out=[]
|
|
for _,r in df.iterrows():
|
|
px=float(r['close'])
|
|
if pos>0:
|
|
peak=max(peak,px)
|
|
if px<=peak*(1-trail): pos=0; cooldown=20
|
|
else:
|
|
cooldown-=1
|
|
if cooldown<=0 and px>=df.loc[:r.name,'close'].rolling(50,min_periods=20).mean().iloc[-1]:
|
|
pos=1; peak=px
|
|
out.append({'date':r['date'],'position':pos})
|
|
return pd.DataFrame(out)
|
|
|
|
|
|
def backtest(df: pd.DataFrame, posdf: pd.DataFrame, cost=ONE_WAY_COST) -> pd.DataFrame:
|
|
z=df[['date','close']].merge(posdf[['date','position']],on='date',how='inner').copy()
|
|
z['asset_return']=z['close'].pct_change().fillna(0)
|
|
z['position_lag']=z['position'].shift(1).fillna(z['position'].iloc[0])
|
|
delta=z['position'].diff().fillna(0)
|
|
z['turnover']=delta.abs()
|
|
z['cost']=z['turnover']*cost
|
|
z['strategy_return']=z['position_lag']*z['asset_return']-z['cost']
|
|
z['equity']=(1+z['strategy_return']).cumprod()
|
|
return z
|
|
|
|
|
|
def max_recovery_days(equity: pd.Series) -> int:
|
|
dd=equity/equity.cummax()-1
|
|
cur=mx=0
|
|
for v in dd<0:
|
|
cur=cur+1 if v else 0; mx=max(mx,cur)
|
|
return int(mx)
|
|
|
|
|
|
def metrics(bt: pd.DataFrame) -> dict:
|
|
if len(bt)<2: return {}
|
|
years=max((bt['date'].iloc[-1]-bt['date'].iloc[0]).days/365.25,1/365.25)
|
|
eq=bt['equity']; ret=bt['strategy_return']
|
|
final=float(eq.iloc[-1]); cagr=final**(1/years)-1
|
|
vol=float(ret.std(ddof=1)*math.sqrt(252)) if ret.std(ddof=1)>0 else 0
|
|
sh=float(ret.mean()/ret.std(ddof=1)*math.sqrt(252)) if ret.std(ddof=1)>0 else 0
|
|
dd=eq/eq.cummax()-1; mdd=float(dd.min())
|
|
downside=ret[ret<0]
|
|
sortino=float(ret.mean()/downside.std(ddof=1)*math.sqrt(252)) if len(downside)>2 and downside.std(ddof=1)>0 else 0
|
|
cvar=float(ret[ret<=ret.quantile(.05)].mean()) if len(ret)>20 else np.nan
|
|
return {
|
|
'start':bt['date'].iloc[0].date().isoformat(),'end':bt['date'].iloc[-1].date().isoformat(),'observations':len(bt),
|
|
'cagr':cagr,'mdd':mdd,'calmar':cagr/abs(mdd) if mdd<0 else np.nan,'annual_vol':vol,'sharpe_zero_rf':sh,'sortino_zero_rf':sortino,
|
|
'ulcer_index':float(np.sqrt(np.mean((dd*100)**2))),'recovery_days_max':max_recovery_days(eq),
|
|
'turnover_annual':float(bt['turnover'].sum()/years),'avg_exposure':float(bt['position_lag'].mean()),
|
|
'cvar_5_daily':cvar,'final_equity':final,
|
|
}
|
|
|
|
|
|
def event_stats(symbol: str, df: pd.DataFrame, events: pd.DataFrame, strategy: str) -> dict:
|
|
if events.empty:
|
|
return {'symbol':symbol,'strategy':strategy,'sell_events':0,'buy_events':0,'median_gain_capture':np.nan,'median_63d_post_sell_return':np.nan,'false_exit_rate_63d':np.nan}
|
|
close=df.set_index('date')['close']
|
|
sells=events[events['side']=='SELL'].copy(); buys=events[events['side']=='BUY']
|
|
captures=[]; fwd=[]; false=[]
|
|
for _,e in sells.iterrows():
|
|
d=e['date']; px=e['price']; entry=e['cycle_entry']; peak=e['peak']
|
|
denom=peak-entry
|
|
captures.append((px-entry)/denom if denom>0 else np.nan)
|
|
idx=close.index.get_indexer([d])[0]
|
|
if idx>=0 and idx+63<len(close):
|
|
r=close.iloc[idx+63]/px-1; fwd.append(r); false.append(r>0.10)
|
|
return {'symbol':symbol,'strategy':strategy,'sell_events':len(sells),'buy_events':len(buys),
|
|
'median_gain_capture':float(np.nanmedian(captures)) if captures else np.nan,
|
|
'median_63d_post_sell_return':float(np.nanmedian(fwd)) if fwd else np.nan,
|
|
'false_exit_rate_63d':float(np.mean(false)) if false else np.nan}
|
|
|
|
|
|
def period_slice(bt, start, end):
|
|
q=bt[(bt['date']>=pd.Timestamp(start)) & (bt['date']<=pd.Timestamp(end))].copy()
|
|
if len(q):
|
|
# Rebase within period.
|
|
q['equity']=(1+q['strategy_return']).cumprod()
|
|
return q
|
|
|
|
|
|
def portfolio_backtest(asset_bts: Dict[str,pd.DataFrame], use_capital_floor: bool) -> pd.DataFrame:
|
|
rets=[]; exps=[]; turns=[]
|
|
for s,bt in asset_bts.items():
|
|
q=bt.set_index('date')
|
|
rets.append(q['strategy_return'].rename(s)); exps.append(q['position_lag'].rename(s)); turns.append(q['turnover'].rename(s))
|
|
R=pd.concat(rets,axis=1).sort_index()
|
|
E=pd.concat(exps,axis=1).reindex(R.index)
|
|
T=pd.concat(turns,axis=1).reindex(R.index)
|
|
base=R.mean(axis=1,skipna=True).fillna(0)
|
|
avgexp=E.mean(axis=1,skipna=True).fillna(0)
|
|
avgturn=T.mean(axis=1,skipna=True).fillna(0)
|
|
scale=1.0; equity=1.0; peak=1.0; prev_scale=1.0; rows=[]
|
|
for d,r in base.items():
|
|
dd=equity/peak-1
|
|
target=1.0
|
|
if use_capital_floor:
|
|
if dd<=-.25: target=.35
|
|
elif dd<=-.18: target=.55
|
|
elif dd<=-.12: target=.75
|
|
elif dd<=-.08: target=.90
|
|
# Hysteresis: scale up only in 10%-points per day after recovery.
|
|
if target>scale: scale=min(target,scale+.10)
|
|
else: scale=target
|
|
else: scale=1.0
|
|
overlay_turn=abs(scale-prev_scale)
|
|
strat=scale*r-overlay_turn*ONE_WAY_COST
|
|
equity*=1+strat; peak=max(peak,equity)
|
|
rows.append({'date':d,'asset_return':r,'position':scale*avgexp.loc[d],'position_lag':scale*avgexp.loc[d],
|
|
'turnover':scale*avgturn.loc[d]+overlay_turn,'cost':overlay_turn*ONE_WAY_COST,'strategy_return':strat,'equity':equity,'risk_scale':scale})
|
|
prev_scale=scale
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def main():
|
|
dfs={s:load_symbol(s,p) for s,p in FILES.items()}
|
|
pd.DataFrame([data_quality(s,df) for s,df in dfs.items()]).to_csv(OUT/'data_quality.csv',index=False)
|
|
asset_rows=[]; period_rows=[]; event_rows=[]
|
|
bts_by_strategy={k:{} for k in ['buy_hold','trend_10m_shifted','trailing_12','kartsell_v11_original','kartsell_v12_candidate']}
|
|
periods={
|
|
'research_2007_2013':('2007-01-01','2013-12-31'),
|
|
'validation_2014_2017':('2014-01-01','2017-12-31'),
|
|
'rolling_oos_2018_2023':('2018-01-01','2023-12-31'),
|
|
'frozen_oos_2024_latest':('2024-01-01','2026-08-01'),
|
|
'gfc_2007_2009':('2007-01-01','2009-12-31'),
|
|
'covid_2020':('2020-01-01','2020-12-31'),
|
|
'inflation_bear_2022':('2022-01-01','2022-12-31'),
|
|
}
|
|
for s,df in dfs.items():
|
|
v11,e11=v11_positions(df,ASSET_CLASS[s]); v12,e12=v12_positions(df,ASSET_CLASS[s])
|
|
posmap={'buy_hold':buyhold_positions(df),'trend_10m_shifted':trend10m_positions(df),'trailing_12':trailing_positions(df),
|
|
'kartsell_v11_original':v11,'kartsell_v12_candidate':v12}
|
|
for name,pos in posmap.items():
|
|
bt=backtest(df,pos); bts_by_strategy[name][s]=bt
|
|
asset_rows.append({'symbol':s,'asset_class':ASSET_CLASS[s],'strategy':name,**metrics(bt)})
|
|
for pn,(ps,pe) in periods.items():
|
|
q=period_slice(bt,ps,pe)
|
|
if len(q)>=60:
|
|
period_rows.append({'symbol':s,'strategy':name,'period':pn,**metrics(q)})
|
|
e11.to_csv(OUT/f'{s}_v11_events.csv',index=False); e12.to_csv(OUT/f'{s}_v12_events.csv',index=False)
|
|
event_rows += [event_stats(s,df,e11,'kartsell_v11_original'),event_stats(s,df,e12,'kartsell_v12_candidate')]
|
|
pd.DataFrame(asset_rows).to_csv(OUT/'asset_metrics.csv',index=False)
|
|
pd.DataFrame(period_rows).to_csv(OUT/'period_metrics.csv',index=False)
|
|
pd.DataFrame(event_rows).to_csv(OUT/'sell_event_stats.csv',index=False)
|
|
|
|
port_rows=[]; port_bts={}
|
|
portfolio_sets = {
|
|
'core_six_dynamic_equal_weight': ['SPY','AAPL','AMZN','NVDA','TSLA','AIG'],
|
|
'all_seven_including_delisted_stress': list(FILES.keys()),
|
|
}
|
|
for portfolio_name, symbols in portfolio_sets.items():
|
|
for name,abd in bts_by_strategy.items():
|
|
selected={k:v for k,v in abd.items() if k in symbols}
|
|
p=portfolio_backtest(selected,use_capital_floor=False)
|
|
port_bts[f'{portfolio_name}_{name}']=p
|
|
port_rows.append({'portfolio':portfolio_name,'strategy':name,**metrics(p)})
|
|
selected={k:v for k,v in bts_by_strategy['kartsell_v12_candidate'].items() if k in symbols}
|
|
pcap=portfolio_backtest(selected,use_capital_floor=True)
|
|
port_bts[f'{portfolio_name}_kartsell_v12_plus_capital_floor']=pcap
|
|
port_rows.append({'portfolio':portfolio_name,'strategy':'kartsell_v12_plus_capital_floor',**metrics(pcap)})
|
|
pd.DataFrame(port_rows).to_csv(OUT/'portfolio_metrics.csv',index=False)
|
|
for name,p in port_bts.items(): p.to_csv(OUT/f'portfolio_{name}.csv',index=False)
|
|
|
|
# Compact summary comparison.
|
|
am=pd.DataFrame(asset_rows)
|
|
wide=am.pivot(index='symbol',columns='strategy',values=['cagr','mdd','calmar','turnover_annual','avg_exposure'])
|
|
wide.to_csv(OUT/'asset_comparison_wide.csv')
|
|
print(pd.DataFrame(port_rows)[['strategy','cagr','mdd','calmar','turnover_annual','avg_exposure','final_equity']].to_string(index=False))
|
|
print('\nAsset v11 vs v12:')
|
|
print(am[am.strategy.isin(['kartsell_v11_original','kartsell_v12_candidate'])][['symbol','strategy','cagr','mdd','calmar','turnover_annual','avg_exposure']].to_string(index=False))
|
|
|
|
if __name__=='__main__':
|
|
main()
|