|
import yfinance as yf |
|
from backtesting import Backtest |
|
import pandas as pd |
|
import os |
|
|
|
from multiprocessing import Pool |
|
from itertools import repeat |
|
from functools import partial |
|
from strategies import SMC_test, SMC_ema, SMCStructure |
|
|
|
def fetch(symbol, period, interval): |
|
df = yf.download(symbol, period=period, interval=interval) |
|
df.columns =df.columns.get_level_values(0) |
|
return df |
|
|
|
def smc_backtest(data, filename, **kwargs): |
|
bt = Backtest(data, SMC_test, cash=kwargs['cash'], commission=kwargs['commission']) |
|
results = bt.run(swing_window=kwargs['swing_hl']) |
|
bt.plot(filename=filename, open_browser=False) |
|
return results |
|
|
|
def smc_ema_backtest(data, filename, **kwargs): |
|
bt = Backtest(data, SMC_ema, cash=kwargs['cash'], commission=kwargs['commission']) |
|
results = bt.run(swing_window=kwargs['swing_hl'], ema1=kwargs['ema1'], ema2=kwargs['ema2'], close_on_crossover=kwargs['cross_close']) |
|
bt.plot(filename=filename, open_browser=False) |
|
return results |
|
|
|
def smc_structure_backtest(data, filename, **kwargs): |
|
bt = Backtest(data, SMCStructure, cash=kwargs['cash'], commission=kwargs['commission']) |
|
results = bt.run(swing_window=kwargs['swing_hl']) |
|
bt.plot(filename=filename, open_browser=False) |
|
return results |
|
|
|
def run_strategy(ticker_symbol, strategy, period, interval, **kwargs): |
|
|
|
retries = 3 |
|
for i in range(retries): |
|
try: |
|
data = fetch(ticker_symbol, period, interval) |
|
except: |
|
raise Exception(f"{ticker_symbol} data fetch failed") |
|
|
|
if len(data) == 0: |
|
if i < retries - 1: |
|
print(f"Attempt{i + 1}: {ticker_symbol} ohlc is empty") |
|
else: |
|
raise Exception(f"{ticker_symbol} ohlc is empty") |
|
else: |
|
break |
|
|
|
filename = f'{ticker_symbol}.html' |
|
|
|
if strategy == "Order Block": |
|
backtest_results = smc_backtest(data, filename, **kwargs) |
|
elif strategy == "Order Block with EMA": |
|
backtest_results = smc_ema_backtest(data, filename, **kwargs) |
|
elif strategy == "Structure trading": |
|
backtest_results = smc_structure_backtest(data, filename, **kwargs) |
|
else: |
|
raise Exception('Strategy not found') |
|
|
|
with open(filename, 'r', encoding='utf-8') as f: |
|
plot = f.read() |
|
|
|
os.remove(filename) |
|
|
|
|
|
backtest_results = backtest_results.to_frame().transpose() |
|
|
|
backtest_results['stock'] = ticker_symbol |
|
backtest_results['plot'] = plot |
|
|
|
|
|
cols = ['stock', 'Start', 'End', 'Return [%]', 'Equity Final [$]', 'Buy & Hold Return [%]', '# Trades', |
|
'Win Rate [%]', 'Best Trade [%]', 'Worst Trade [%]', 'Avg. Trade [%]', 'plot'] |
|
backtest_results = backtest_results[cols] |
|
|
|
backtest_results = backtest_results.rename(columns = {'Equity Final [$]': 'Equity Final [₹]'}) |
|
|
|
return backtest_results |
|
|
|
def complete_test(strategy: str, period: str, interval: str, multiprocess=True, **kwargs): |
|
nifty50 = pd.read_csv("data/ind_nifty50list.csv") |
|
ticker_list = pd.read_csv("data/Ticker_List_NSE_India.csv") |
|
|
|
|
|
nifty50 = nifty50.merge(ticker_list, "inner", left_on=['Symbol'], right_on=['SYMBOL']) |
|
|
|
if multiprocess: |
|
with Pool() as p: |
|
result = p.starmap(partial(run_strategy, **kwargs), zip(nifty50['YahooEquiv'].values, repeat(strategy), repeat(period), repeat(interval))) |
|
else: |
|
result = [run_strategy(nifty50['YahooEquiv'].values[i], strategy, period, interval, **kwargs) for i in range(len(nifty50))] |
|
|
|
df = pd.concat(result) |
|
|
|
df['plot'] = df['plot'].astype(str) |
|
df = df.sort_values(by=['Return [%]'], ascending=False) |
|
|
|
return df.reset_index().drop(columns=['index']) |
|
|
|
|
|
if __name__ == "__main__": |
|
|
|
|
|
|
|
|
|
rt = complete_test("Order Block", '1mo', '15m', swing_hl=20) |
|
rt.to_excel('test/all_testing_2.xlsx', index=False) |
|
print(rt) |