| """Backtest engine""" |
| import pandas as pd, numpy as np |
| from core import Trade, CAPITAL, BROKER, STT, SLIP, MAXPOS, DELAY, MAXTRADES |
| from collections import Counter |
|
|
| def backtest(signals): |
| trades=[]; pos=None; daily_pnl={}; day_trades={}; equity=[CAPITAL] |
| for idx,row in signals.iterrows(): |
| dt=row.get('datetime',idx); date=row.get('date',dt.date() if hasattr(dt,'date') else dt) |
| tm=row.get('time',dt.time() if hasattr(dt,'time') else None) |
| close=float(row['close']); sig=int(row.get('signal',0)) |
| sl=row.get('sl_price',None); tp=row.get('tp_price',None) |
| lo=float(row['low']); hi=float(row['high']) |
| if date not in day_trades: day_trades[date]=0 |
| if pos is not None: |
| ex=False; ep=close; er=""; d=pos['direction'] |
| if d==1 and lo<=pos['sl']: ep=pos['sl']; ex=True; er="stop_loss" |
| elif d==-1 and hi>=pos['sl']: ep=pos['sl']; ex=True; er="stop_loss" |
| elif d==1 and hi>=pos['tp']: ep=pos['tp']; ex=True; er="take_profit" |
| elif d==-1 and lo<=pos['tp']: ep=pos['tp']; ex=True; er="take_profit" |
| elif tm and hasattr(tm,'hour') and tm.hour==15 and tm.minute>=15: ep=close; ex=True; er="eod_exit" |
| elif sig!=0 and sig!=d: ep=close; ex=True; er="signal_exit" |
| if ex: |
| ep=ep*(1-SLIP) if d==1 else ep*(1+SLIP) |
| sh=pos['shares']; gross=(ep-pos['entry_price'])*d*sh |
| costs=BROKER+pos['entry_price']*sh*STT+BROKER+ep*sh*STT; net=gross-costs |
| trades.append(Trade(pos['ticker'],pos['entry_time'],dt,d,pos['entry_price'],ep,sh,net, |
| net/(pos['entry_price']*sh) if pos['entry_price']*sh!=0 else 0,0,er)) |
| equity.append(equity[-1]+net) |
| daily_pnl[date]=daily_pnl.get(date,0)+net; day_trades[date]+=1; pos=None |
| if pos is None and sig!=0: |
| if day_trades.get(date,0)<MAXTRADES: |
| if row.get('bar_num',999)>=DELAY: |
| ep=close*(1+SLIP) if sig==1 else close*(1-SLIP) |
| sh=int(CAPITAL*MAXPOS/ep) |
| if sh>0: |
| dsl=ep*(1-0.008*sig); dtp=ep*(1+0.012*sig) |
| psl=float(sl) if sl is not None and not pd.isna(sl) else dsl |
| ptp=float(tp) if tp is not None and not pd.isna(tp) else dtp |
| pos={'ticker':row.get('ticker',''),'entry_time':dt,'entry_price':ep, |
| 'direction':sig,'shares':sh,'sl':psl,'tp':ptp} |
| return trades, daily_pnl, equity |
|
|
| def metrics(name, all_trades, all_daily): |
| if not all_trades: return None |
| pnls=[t.pnl for t in all_trades] |
| w=[t for t in all_trades if t.pnl>0]; l=[t for t in all_trades if t.pnl<=0] |
| tr=sum(pnls)/CAPITAL |
| ds=pd.Series(all_daily).sort_index(); dr=ds/CAPITAL |
| sh=(dr.mean()/dr.std()*np.sqrt(252)) if len(dr)>1 and dr.std()>0 else 0 |
| dd=dr[dr<0]; so=(dr.mean()/dd.std()*np.sqrt(252)) if len(dd)>0 and dd.std()>0 else 0 |
| eq=CAPITAL+ds.cumsum(); mdd=abs(((eq-eq.cummax())/eq.cummax()).min()) if len(eq)>0 else 0 |
| gp=sum(t.pnl for t in w) if w else 0; gl=abs(sum(t.pnl for t in l)) if l else 1 |
| pf=gp/gl if gl>0 else 0 |
| ex=Counter(t.exit_reason for t in all_trades); sp={} |
| for t in all_trades: sp[t.ticker]=sp.get(t.ticker,0)+t.pnl |
| return dict(strategy=name, total_trades=len(all_trades), win_rate=len(w)/len(all_trades)*100, |
| total_return_pct=tr*100, sharpe_ratio=sh, sortino_ratio=so, max_drawdown_pct=mdd*100, |
| profit_factor=pf, avg_pnl=np.mean(pnls), avg_win=np.mean([t.pnl for t in w]) if w else 0, |
| avg_loss=np.mean([t.pnl for t in l]) if l else 0, trades=all_trades, exits=ex, stock_pnl=sp) |
|
|
