import os os.environ.setdefault('MPLBACKEND', 'Agg') from pathlib import Path from dotenv import load_dotenv load_dotenv(Path.home() / '.env') import json, itertools, contextlib, io import numpy as np import pandas as pd import finlab from finlab import data from finlab.backtest import sim from strategy import load_prices, make_position, START, END ROOT = Path(__file__).parent OUT = ROOT / 'results' OUT.mkdir(exist_ok=True) SLUGS = {'hs':'head-and-shoulders-backtest', 'wm':'double-bottom-top-backtest', 'fib':'fibonacci-retracement-backtest', 'cup':'cup-and-handle-backtest'} SHAPE_GRID = [dict(bandwidth=h, tolerance=t) for h,t in itertools.product([2,3,4],[.02,.03,.04])] GRIDS = {'hs': SHAPE_GRID, 'wm': SHAPE_GRID, 'fib': [dict(bandwidth=h, window=w) for h,w in itertools.product([2,3,4],[60,90,120])], 'cup': [dict(cup_depth=d, handle_depth=h) for d,h in itertools.product([.25,.33,.4],[.08,.12,.16])]} CENTRAL = 4 # middle of each 3x3 grid def daily_returns(prices, index): return prices.reindex(index).ffill().pct_change(fill_method=None).fillna(0) def daily_metrics(returns): r = returns.fillna(0) sd = r.std(ddof=1) wealth = (1+r).cumprod() return dict(cagr=float(wealth.iloc[-1] ** (365.25 / max((r.index[-1] - r.index[0]).days, 1))-1), sharpe=float(r.mean()/sd*np.sqrt(252)) if sd > 0 else 0., mdd=float((wealth / wealth.cummax()-1).min()), vol=float(sd*np.sqrt(252))) all_results, details = [], {} for market in ['tw', '0050', 'us']: print('LOAD', market, flush=True) prices = load_prices(market) close, eligible = prices[0], prices[-1] print('SHAPE', close.shape, 'eligible', int(eligible.sum().sum()), flush=True) if market == 'us': funds = pd.DataFrame(data.get('us_fund_price:adj_close')) benchmark = funds['QQQ'] bil = funds['BIL'] else: benchmark = pd.DataFrame(data.get('etl:adj_close'))['0050'] bil = None index = close.loc[START:END].index bench_r = daily_returns(benchmark, index) bil_r = daily_returns(bil, index) if bil is not None else bench_r * 0 stock_r = daily_returns(close, index) midpoint = len(index) // 2 limit = 1 if market == '0050' else .1 kwargs = dict(position_limit=limit, upload=False, end_date=END) bench_stats = daily_metrics(bench_r) bench_halves = [daily_metrics(x)['sharpe'] for x in [bench_r.iloc[:midpoint],bench_r.iloc[midpoint:]]] bench_excess_sharpe = daily_metrics(bench_r-bil_r)['sharpe'] families = ['cup', 'hs'] if market == 'us' else ['hs', 'wm', 'fib'] for family in families: ratios = [.382, .5, .618] if family == 'fib' else [.5] for ratio, inverse in itertools.product(ratios, [False, True]): key = f'{market}_{family}_{ratio}_{"avoid" if inverse else "long"}' rows = [] for i, params in enumerate(GRIDS[family]): position, entries = make_position(prices, family=family, ratio=ratio, inverse=inverse, **params) with contextlib.redirect_stdout(io.StringIO()): report = sim(position, **kwargs) equity = report.creturn.reindex(index).ffill().fillna(1) returns = equity.pct_change(fill_method=None).fillna(0) weights = position.astype(float) * limit mean_weights = weights.mean() fixed_r = stock_r.mul(mean_weights).sum(axis=1) daily_exposure = weights.sum(axis=1) exposure = daily_exposure.mean() constant_exposure_r = bench_r * exposure stats = daily_metrics(returns) turnover = weights.diff().abs().sum(axis=1) / 2 halves = [daily_metrics(x)['sharpe'] for x in [returns.iloc[:midpoint],returns.iloc[midpoint:]]] annual = returns.groupby(returns.index.year).apply(lambda r: (1+r).prod()-1) best_year = int(annual.idxmax()) without = returns.index.year != best_year cost_stats = None cost_pass = True if market == 'us': with contextlib.redirect_stdout(io.StringIO()): cost_report = sim(position, fee_ratio=.001, **kwargs) cost_returns = cost_report.creturn.reindex(index).ffill().fillna(1).pct_change(fill_method=None).fillna(0) cost_stats = daily_metrics(cost_returns) cost_pass = cost_stats['sharpe'] > bench_stats['sharpe'] row = dict(design=key, market=market, family=family, ratio=ratio, inverse=inverse, grid=i, parameters=params, **stats, benchmark=bench_stats, half_sharpe=halves, benchmark_half_sharpe=bench_halves, exposure=float(exposure), signal_count=int((entries.loc[START:END] & eligible.loc[START:END]).sum().sum()), selected_entries=int((position & ~position.shift(1).fillna(False)).sum().sum()), fixed=daily_metrics(fixed_r), constant_exposure=daily_metrics(constant_exposure_r), excess_sharpe=daily_metrics(returns-bil_r)['sharpe'], benchmark_excess_sharpe=bench_excess_sharpe, best_year=best_year, without_best=daily_metrics(returns.loc[without])['sharpe'], benchmark_without_best=daily_metrics(bench_r.loc[without])['sharpe'], cost=cost_stats, cost_pass=cost_pass, turnover_mean=float(turnover.mean()), turnover_month_max=float(turnover.resample('ME').sum().max())) row['wins'] = stats['sharpe'] > row['benchmark']['sharpe'] rows.append(row) all_results.append(row) if i == CENTRAL: folder = OUT / SLUGS[family] folder.mkdir(exist_ok=True) with contextlib.redirect_stdout(io.StringIO()): report.to_html(str(folder / f'{key}_report.html')) pd.DataFrame({'strategy':equity/equity.iloc[0], 'benchmark':(1+bench_r).cumprod(), 'fixed_weights':(1+fixed_r).cumprod(), 'constant_exposure':(1+constant_exposure_r).cumprod(), 'exposure': daily_exposure}).to_csv(folder / f'{key}_daily.csv',index_label='date') details[key] = row print('RUN',key,i,round(stats['sharpe'],3),flush=True) central = rows[CENTRAL] central['neighbor_wins'] = sum(r['wins'] for r in rows) central['pass'] = bool(central['neighbor_wins']>=7 and all(a>b for a,b in zip(central['half_sharpe'],central['benchmark_half_sharpe'])) and central['wins'] and central['cost_pass'] and (central['exposure']>=1-1e-8 or central['sharpe']>central['fixed']['sharpe']) and central['sharpe']>central['constant_exposure']['sharpe'] and central['excess_sharpe']>central['benchmark_excess_sharpe'] and central['without_best']>central['benchmark_without_best']) print('DESIGN',key,'wins',central['neighbor_wins'],'PASS',central['pass'],flush=True) (OUT/'grid.json').write_text(json.dumps(all_results,indent=2)) (OUT/'central.json').write_text(json.dumps(details,indent=2)) del prices, close, eligible metadata = dict(run_date=str(pd.Timestamp.now(tz='Asia/Taipei').date()), finlab_version=finlab.__version__, start=START,end=END,preregistration='https://github.com/finlab-python/finlab-website/issues/448#issuecomment-6782115216') (OUT/'metadata.json').write_text(json.dumps(metadata,indent=2))