218 lines
11 KiB
Python
218 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import math
|
|
from pathlib import Path
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
import sys
|
|
sys.path.insert(0, '/mnt/data')
|
|
import kartsell_v12_research as base
|
|
|
|
ROOT = Path('/mnt/data')
|
|
OUT = ROOT / 'kartsell_v12_2_results'
|
|
OUT.mkdir(exist_ok=True)
|
|
|
|
KR_FILES = {
|
|
'005930': ROOT / '005930_KR.csv',
|
|
'000660': ROOT / '000660_KR.csv',
|
|
'035900': ROOT / '035900_KR.csv',
|
|
}
|
|
KR_NAMES = {'005930':'Samsung Electronics','000660':'SK Hynix','035900':'JYP Ent.'}
|
|
KR_CLASS = {'005930':'large_quality_stock','000660':'cyclical_stock','035900':'high_vol_stock'}
|
|
|
|
US_FILES = base.FILES
|
|
US_CLASS = base.ASSET_CLASS
|
|
|
|
|
|
def load_kr(symbol: str, path: Path) -> pd.DataFrame:
|
|
raw = pd.read_csv(path)
|
|
raw.columns = [c.strip().lower() for c in raw.columns]
|
|
raw['date'] = pd.to_datetime(raw['date'], errors='coerce')
|
|
for c in ['open','high','low','close','volume','adj_close']:
|
|
raw[c] = pd.to_numeric(raw[c], errors='coerce')
|
|
raw = raw.dropna(subset=['date','open','high','low','close','adj_close']).copy()
|
|
raw = raw[(raw['date'] >= pd.Timestamp('2007-01-01')) & (raw['date'] <= pd.Timestamp('2026-08-01'))]
|
|
raw = raw.sort_values('date').drop_duplicates('date', keep='last').reset_index(drop=True)
|
|
|
|
# Historical Korean Yahoo files sometimes put raw Close on a pre-split scale while OHLC is already adjusted.
|
|
# If raw Close is inside the raw daily range, OHLC and Close share a scale and all OHLC are adjusted by adj/close.
|
|
# If Adj_Close is inside the range but raw Close is not, the OHLC is already on the adjusted scale (notably old JYP rows).
|
|
raw = raw[(raw['adj_close'] > 0) & (raw['open'] > 0) & (raw['high'] > 0) & (raw['low'] > 0) & (raw['close'] > 0)].copy()
|
|
raw_in_range = (raw['close'] >= raw['low'] * .995) & (raw['close'] <= raw['high'] * 1.005)
|
|
adj_in_range = (raw['adj_close'] >= raw['low'] * .995) & (raw['adj_close'] <= raw['high'] * 1.005)
|
|
scale_raw = raw_in_range | (~adj_in_range)
|
|
factor = raw['adj_close'] / raw['close']
|
|
for c in ['open','high','low']:
|
|
raw[c] = np.where(scale_raw, raw[c] * factor, raw[c])
|
|
raw['close'] = raw['adj_close']
|
|
|
|
# Infer only near-integer discontinuities above 80% as provisional corporate-action adjustments.
|
|
# These are audit flags, not a substitute for an official corporate-action master.
|
|
raw = raw.reset_index(drop=True)
|
|
inferred = []
|
|
for i in range(1, len(raw)):
|
|
prev = float(raw.at[i-1,'close']); cur = float(raw.at[i,'close'])
|
|
if prev <= 0 or cur <= 0:
|
|
continue
|
|
ratio = cur / prev
|
|
mult = None
|
|
if ratio >= 1.8:
|
|
k = round(ratio)
|
|
if 2 <= k <= 100 and abs(ratio-k)/k <= .04:
|
|
mult = float(k)
|
|
elif ratio <= .55:
|
|
inv = 1.0/ratio
|
|
k = round(inv)
|
|
if 2 <= k <= 100 and abs(inv-k)/k <= .04:
|
|
mult = 1.0/float(k)
|
|
if mult is not None:
|
|
raw.loc[:i-1, ['open','high','low','close','adj_close']] *= mult
|
|
inferred.append((raw.at[i,'date'], mult))
|
|
raw.attrs['inferred_actions'] = inferred
|
|
raw['symbol'] = symbol
|
|
return raw[['date','symbol','open','high','low','close','adj_close','volume']]
|
|
|
|
|
|
def quality(symbol: str, path: Path, df: pd.DataFrame) -> dict:
|
|
raw = pd.read_csv(path)
|
|
dates = pd.to_datetime(raw['Date'], errors='coerce')
|
|
ret = df['close'].pct_change()
|
|
invalid = ((df['high'] < df[['open','close','low']].max(axis=1)) |
|
|
(df['low'] > df[['open','close','high']].min(axis=1))).sum()
|
|
return {
|
|
'symbol':symbol,'name':KR_NAMES[symbol],'market':'KOSPI' if symbol!='035900' else 'KOSDAQ',
|
|
'start':df.date.min().date().isoformat(),'end':df.date.max().date().isoformat(),'rows':len(df),
|
|
'raw_sorted':bool(dates.dropna().is_monotonic_increasing),'duplicate_dates':int(dates.duplicated().sum()),
|
|
'invalid_adjusted_ohlc_rows':int(invalid),'abs_daily_return_gt_50pct':int((ret.abs()>.5).sum()),
|
|
'zero_or_missing_volume_rows':int((df.volume.fillna(0)<=0).sum()),
|
|
'total_return_available':False,'pit_fundamentals_available':False,
|
|
'inferred_corporate_action_rescales': {'005930':'1:50 pre-2018-04-09','035900':'2:1 pre-2007-10-04'}.get(symbol,''),
|
|
'source_vintage_warning':'source ends 2021-04; not current'
|
|
}
|
|
|
|
|
|
def run_assets(dfs: dict[str,pd.DataFrame], classes: dict[str,str], market: str, cost: float):
|
|
rows=[]; periods=[]; events=[]; bts={k:{} for k in ['buy_hold','trend_10m_shifted','trailing_12','kartsell_v11_original','kartsell_v12_candidate']}
|
|
period_def={
|
|
'research_2007_2013':('2007-01-01','2013-12-31'),
|
|
'validation_2014_2017':('2014-01-01','2017-12-31'),
|
|
'oos_2018_source_end':('2018-01-01','2026-08-01'),
|
|
'gfc_2007_2009':('2007-01-01','2009-12-31'),
|
|
'covid_2020':('2020-01-01','2020-12-31'),
|
|
}
|
|
for s,df in dfs.items():
|
|
v11,e11=base.v11_positions(df,classes[s]); v12,e12=base.v12_positions(df,classes[s])
|
|
posmap={'buy_hold':base.buyhold_positions(df),'trend_10m_shifted':base.trend10m_positions(df),
|
|
'trailing_12':base.trailing_positions(df),'kartsell_v11_original':v11,'kartsell_v12_candidate':v12}
|
|
for n,p in posmap.items():
|
|
bt=base.backtest(df,p,cost=cost); bts[n][s]=bt
|
|
rows.append({'market':market,'symbol':s,'strategy':n,'cost_one_way':cost,**base.metrics(bt)})
|
|
for pn,(ps,pe) in period_def.items():
|
|
q=base.period_slice(bt,ps,pe)
|
|
if len(q)>=60: periods.append({'market':market,'symbol':s,'strategy':n,'period':pn,**base.metrics(q)})
|
|
events.extend([base.event_stats(s,df,e11,'kartsell_v11_original'),base.event_stats(s,df,e12,'kartsell_v12_candidate')])
|
|
e11.to_csv(OUT/f'{s}_v11_events.csv',index=False); e12.to_csv(OUT/f'{s}_v12_events.csv',index=False)
|
|
return pd.DataFrame(rows),pd.DataFrame(periods),pd.DataFrame(events),bts
|
|
|
|
|
|
def portfolio_from(bts, name: str, cost: float):
|
|
rows=[]; curves={}
|
|
for strategy, assets in bts.items():
|
|
p=base.portfolio_backtest(assets,use_capital_floor=False)
|
|
curves[strategy]=p; rows.append({'portfolio':name,'strategy':strategy,'cost_one_way':cost,**base.metrics(p)})
|
|
p=base.portfolio_backtest(bts['kartsell_v12_candidate'],use_capital_floor=True)
|
|
curves['kartsell_v12_plus_capital_floor']=p
|
|
rows.append({'portfolio':name,'strategy':'kartsell_v12_plus_capital_floor','cost_one_way':cost,**base.metrics(p)})
|
|
return pd.DataFrame(rows),curves
|
|
|
|
|
|
def factor_panel(dfs: dict[str,pd.DataFrame], market_map: dict[str,str]) -> pd.DataFrame:
|
|
frames=[]
|
|
for s,df in dfs.items():
|
|
x=df[['date','close']].copy().set_index('date')
|
|
m=x['close'].resample('ME').last().dropna().to_frame('close')
|
|
m['mom12_1']=m['close'].shift(1)/m['close'].shift(12)-1
|
|
m['mom6_1']=m['close'].shift(1)/m['close'].shift(6)-1
|
|
daily=df.set_index('date')['close'].pct_change()
|
|
vol63=daily.rolling(63,min_periods=40).std()*math.sqrt(252)
|
|
high252=df.set_index('date')['close'].rolling(252,min_periods=126).max()
|
|
m['vol63']=vol63.resample('ME').last()
|
|
m['drawdown252']=(df.set_index('date')['close']/high252-1).resample('ME').last()
|
|
m['fwd12m']=m['close'].shift(-12)/m['close']-1
|
|
m['symbol']=s; m['market']=market_map[s]
|
|
frames.append(m.reset_index())
|
|
return pd.concat(frames,ignore_index=True).dropna(subset=['fwd12m'])
|
|
|
|
|
|
def factor_summary(panel: pd.DataFrame) -> pd.DataFrame:
|
|
rows=[]
|
|
# Monthly cross-sectional terciles within region; no future data in factor formation.
|
|
for fac,ascending in [('mom12_1',False),('mom6_1',False),('vol63',True),('drawdown252',False)]:
|
|
p=panel.dropna(subset=[fac]).copy()
|
|
p['rank_pct']=p.groupby(['market','date'])[fac].rank(pct=True,ascending=True)
|
|
# For momentum/drawdown, high factor is preferred; for vol, low preferred.
|
|
if fac=='vol63':
|
|
top=p[p.rank_pct<=1/3]; bottom=p[p.rank_pct>=2/3]
|
|
else:
|
|
top=p[p.rank_pct>=2/3]; bottom=p[p.rank_pct<=1/3]
|
|
for label,q in [('preferred_tercile',top),('opposite_tercile',bottom)]:
|
|
r=q['fwd12m']
|
|
rows.append({'factor':fac,'bucket':label,'observations':len(q),'mean_fwd12m':r.mean(),'median_fwd12m':r.median(),
|
|
'positive_rate':(r>0).mean(),'loss_gt_20_rate':(r<-.20).mean()})
|
|
rows.append({'factor':fac,'bucket':'preferred_minus_opposite','observations':min(len(top),len(bottom)),
|
|
'mean_fwd12m':top.fwd12m.mean()-bottom.fwd12m.mean(),
|
|
'median_fwd12m':top.fwd12m.median()-bottom.fwd12m.median(),
|
|
'positive_rate':np.nan,'loss_gt_20_rate':np.nan})
|
|
return pd.DataFrame(rows)
|
|
|
|
|
|
def core_tactical_spy(spy: pd.DataFrame):
|
|
v12,_=base.v12_positions(spy,'broad_index_etf')
|
|
out=[]
|
|
for core in [.50,.65,.80]:
|
|
p=v12[['date','position']].copy(); p['position']=core+(1-core)*p['position']
|
|
bt=base.backtest(spy,p,cost=base.ONE_WAY_COST)
|
|
out.append({'strategy':f'SPY_core_{int(core*100)}_tactical_{int((1-core)*100)}',**base.metrics(bt)})
|
|
return pd.DataFrame(out)
|
|
|
|
|
|
def main():
|
|
kr={s:load_kr(s,p) for s,p in KR_FILES.items()}
|
|
pd.DataFrame([quality(s,KR_FILES[s],d) for s,d in kr.items()]).to_csv(OUT/'kr_data_quality.csv',index=False)
|
|
|
|
kr_rows,kr_periods,kr_events,kr_bts=run_assets(kr,KR_CLASS,'KR',cost=.0010)
|
|
kr_rows.to_csv(OUT/'kr_asset_metrics_10bps.csv',index=False)
|
|
kr_periods.to_csv(OUT/'kr_period_metrics_10bps.csv',index=False)
|
|
kr_events.to_csv(OUT/'kr_sell_event_stats.csv',index=False)
|
|
kr_port,kr_curves=portfolio_from(kr_bts,'kr_three_equal_weight',cost=.0010)
|
|
kr_port.to_csv(OUT/'kr_portfolio_metrics_10bps.csv',index=False)
|
|
for n,c in kr_curves.items(): c.to_csv(OUT/f'kr_portfolio_{n}.csv',index=False)
|
|
|
|
# Cost stress for sell-heavy strategies.
|
|
stress=[]
|
|
for cost in [.0007,.0015,.0030]:
|
|
r,_,_,b=run_assets(kr,KR_CLASS,'KR',cost=cost)
|
|
p,_=portfolio_from(b,f'kr_three_equal_weight_cost_{cost:.4f}',cost=cost)
|
|
stress.append(p[p.strategy.isin(['kartsell_v11_original','kartsell_v12_candidate'])])
|
|
pd.concat(stress,ignore_index=True).to_csv(OUT/'kr_cost_sensitivity.csv',index=False)
|
|
|
|
us={s:base.load_symbol(s,p) for s,p in US_FILES.items() if s!='AAMRQ'}
|
|
all_dfs={**us,**kr}
|
|
market_map={**{s:'US' for s in us},**{s:'KR' for s in kr}}
|
|
panel=factor_panel(all_dfs,market_map)
|
|
panel.to_csv(OUT/'price_factor_panel_monthly.csv',index=False)
|
|
factor_summary(panel).to_csv(OUT/'price_factor_summary.csv',index=False)
|
|
core_tactical_spy(us['SPY']).to_csv(OUT/'spy_core_tactical_metrics.csv',index=False)
|
|
|
|
print('KR DATA QUALITY')
|
|
print(pd.read_csv(OUT/'kr_data_quality.csv').to_string(index=False))
|
|
print('\nKR PORTFOLIO')
|
|
print(kr_port[['strategy','cagr','mdd','calmar','turnover_annual','avg_exposure']].to_string(index=False))
|
|
print('\nFACTOR SUMMARY')
|
|
print(pd.read_csv(OUT/'price_factor_summary.csv').to_string(index=False))
|
|
print('\nSPY CORE TACTICAL')
|
|
print(pd.read_csv(OUT/'spy_core_tactical_metrics.csv')[['strategy','cagr','mdd','calmar','turnover_annual','avg_exposure']].to_string(index=False))
|
|
|
|
if __name__=='__main__': main()
|