import csv
from datetime import datetime
import numpy as np
import pytz
from collections import defaultdict
IST = [Link]('Asia/Kolkata')
RSI_PERIOD = 7
RISK = 5
START_BALANCE = 1000
LEVERAGE = 1
MAX_POSITION_VALUE = START_BALANCE * LEVERAGE
START_HOUR = 8
END_HOUR = 23
FILE_1D = 'BTC_USDC_USDC_1d_last_month.csv'
FILE_15M = 'BTC_USDC_USDC_15m_last_month.csv'
def read_csv_ohlcv(filepath, scale=1000):
candles = []
with open(filepath, newline='') as f:
reader = [Link](f)
for row in reader:
[Link]([
int(row['timestamp']),
float(row['open']) / scale,
float(row['high']) / scale,
float(row['low']) / scale,
float(row['close']) / scale,
float(row['volume'])
])
return candles
def compute_rsi(prices, period=RSI_PERIOD):
if len(prices) < period + 1:
return None
deltas = [Link](prices)
ups = [Link](min=0)
downs = -[Link](max=0)
avg_gain = [Link](ups[-period:])
avg_loss = [Link](downs[-period:])
if avg_loss == 0:
return 100
rs = avg_gain / avg_loss
return 100 - 100 / (1 + rs)
def get_rsi_for_candles(candles, period=RSI_PERIOD):
closes = [c[4] for c in candles]
return compute_rsi(closes[-(period + 1):], period)
def get_sentiment(daily_candle):
rsi = get_rsi_for_candles([daily_candle])
if rsi is None:
return None
if rsi > 60:
return 'bullish'
elif rsi < 40:
return 'bearish'
return None
def backtest():
daily_candles = read_csv_ohlcv(FILE_1D, scale=1000)
m15_candles = read_csv_ohlcv(FILE_15M, scale=1000)
daily_dict = {[Link](c[0] / 1000, IST).date(): c for c in
daily_candles}
m15_days = defaultdict(list)
for c in m15_candles:
dt = [Link](c[0] / 1000, IST)
m15_days[[Link]()].append(c)
balance = START_BALANCE
results = []
for day in sorted(m15_days.keys()):
if day not in daily_dict:
[Link]({'date': [Link]('%Y-%m-%d'), 'trade_num': '',
'time': '', 'sentiment': 'No daily candle',
'entry': '', 'sl': '', 'tp': '', 'qty': '',
'direction': '', 'result': '', 'pnl': 0, 'balance': balance})
continue
sentiment = get_sentiment(daily_dict[day])
if sentiment is None:
[Link]({'date': [Link]('%Y-%m-%d'), 'trade_num': '',
'time': '', 'sentiment': 'No sentiment',
'entry': '', 'sl': '', 'tp': '', 'qty': '',
'direction': '', 'result': '', 'pnl': 0, 'balance': balance})
continue
candles = m15_days[day]
trade_num = 0
i = RSI_PERIOD + 1
while i < len(candles):
slice_candles = candles[i - RSI_PERIOD - 1:i + 1]
rsi_15m = get_rsi_for_candles(slice_candles)
ts = [Link](candles[i][0] / 1000, IST)
if [Link] < START_HOUR or [Link] > END_HOUR:
i += 1
continue
direction = None
sl = None
if sentiment == 'bullish' and rsi_15m and rsi_15m > 60:
direction = 'long'
sl = min(c[3] for c in candles[i - RSI_PERIOD:i + 1])
elif sentiment == 'bearish' and rsi_15m and rsi_15m < 40:
direction = 'short'
sl = max(c[2] for c in candles[i - RSI_PERIOD:i + 1])
if direction:
entry = candles[i][4]
trade_num += 1
qty = min(RISK / abs(entry - sl), MAX_POSITION_VALUE / entry)
tp = entry + abs(entry - sl) if direction == 'long' else entry -
abs(entry - sl)
exit_result = None
exit_price = None
exit_time = None
for j in range(i + 1, len(candles)):
price = candles[j][4]
t = [Link](candles[j][0] / 1000,
IST).strftime('%H:%M')
if direction == 'long':
if price <= sl:
exit_result, exit_price, exit_time = 'SL', price, t
break
elif price >= tp:
exit_result, exit_price, exit_time = 'TP', price, t
break
else:
if price >= sl:
exit_result, exit_price, exit_time = 'SL', price, t
break
elif price <= tp:
exit_result, exit_price, exit_time = 'TP', price, t
break
pnl = RISK if exit_result == 'TP' else -RISK if exit_result == 'SL'
else 0
balance += pnl
[Link]({
'date': [Link]('%Y-%m-%d'),
'trade_num': trade_num,
'time': [Link]('%H:%M'),
'sentiment': sentiment,
'entry': entry,
'sl': sl,
'tp': tp,
'qty': qty,
'direction': direction,
'result': exit_result,
'pnl': pnl,
'balance': balance
})
i = j
else:
i += 1
if trade_num == 0:
[Link]({'date': [Link]('%Y-%m-%d'), 'trade_num': '',
'time': '', 'sentiment': sentiment,
'entry': '', 'sl': '', 'tp': '', 'qty': '',
'direction': '', 'result': 'No trades', 'pnl': 0,
'balance': balance})
with open('backtest_results.csv', 'w', newline='') as f:
writer = [Link](f,
fieldnames=['date', 'trade_num', 'time',
'sentiment', 'entry', 'sl', 'tp', 'qty', 'direction',
'result', 'pnl', 'balance'])
[Link]()
for row in results:
[Link](row)
print('Backtest complete. Results saved to backtest_results.csv')
if __name__ == "__main__":
backtest()