Strategy Logic (Summary)
- Trend filter: We only look for long positions when Close > EMA200 (the main trend is up).
- Entry triggers:
- RSI(14) crosses from below 30 to above 30 (oversold recovery)
- MACD line crosses the signal upward (momentum confirmation)
- Stop/TP: Initial stop is 2×ATR, profit target is 4×ATR (RR ≈ 2). If the price moves favorably ≥1×ATR, the stop is moved to the entry price (simple trailing).
- Risk management: 1% of the capital is risked on each trade (position size based on ATR).
- Costs: Default 0.1% commission and 0.05% slippage.
How to Customize?
- Timeframe: You can set RESAMPLE_RULE = “4H” → “1H” or “1D”.
- Risk/Stop/TP: RISK_PER_TRADE, ATR_MULT_STOP, ATR_MULT_TP.
- Filters & Triggers: Change the RSI threshold (30/70), MACD parameters, trend filter, or add Bollinger/ADX.
- Symbols: We use BTC-USD from Yahoo instead of Binance spot data; try ETH-USD, etc. if you prefer.
🔹1. Imports + Parameters
# pip install yfinance pandas numpy
import math
import pandas as pd
import numpy as np
import yfinance as yf
from datetime import datetime, timezone
# -----------------------------
# Parameters
# -----------------------------
SYMBOL = "BTC-USD" # Yahoo Finance symbol
BASE_INTERVAL = "1h" # Fetch 1h data and resample to 4H
RESAMPLE_RULE = "4H" # Working timeframe
PERIOD = "730d" # Last 2 years
FEE_RATE = 0.001 # 0.1% commission per trade
SLIPPAGE = 0.0005 # 0.05% slippage per trade
RISK_PER_TRADE = 0.01 # Risk 1% of equity per trade
INITIAL_EQUITY = 10000.0 # Starting equity
ATR_MULT_STOP = 2.0 # Stop loss = 2*ATR
ATR_MULT_TP = 4.0 # Take profit = 4*ATR (RR = 2)
RSI_LEN = 14
EMA_TREND_LEN = 200
🔹 2. Indicator Helpers
# -----------------------------
# Indicator helper functions
# -----------------------------
def ema(series, span):
return series.ewm(span=span, adjust=False).mean()
def rsi(series, length=14):
delta = series.diff()
gain = delta.clip(lower=0).ewm(alpha=1/length, adjust=False).mean()
loss = (-delta.clip(upper=0)).ewm(alpha=1/length, adjust=False).mean()
rs = gain / loss
return 100 - (100 / (1 + rs))
def atr(df, length=14):
high, low, close = df['High'], df['Low'], df['Close']
prev_close = close.shift(1)
tr = pd.concat([
(high - low).abs(),
(high - prev_close).abs(),
(low - prev_close).abs()
], axis=1).max(axis=1)
return tr.ewm(alpha=1/length, adjust=False).mean()
🔹 3. Data Download & Indicators
# -----------------------------
# Data download and resample
# -----------------------------
raw = yf.download(SYMBOL, period=PERIOD, interval=BASE_INTERVAL, auto_adjust=False, progress=False)
if raw.empty:
raise SystemExit("No data fetched. Check symbol/period/interval.")
# Resample to 4H (OHLCV)
df = pd.DataFrame()
df['Open'] = raw['Open'].resample(RESAMPLE_RULE).first()
df['High'] = raw['High'].resample(RESAMPLE_RULE).max()
df['Low'] = raw['Low'].resample(RESAMPLE_RULE).min()
df['Close'] = raw['Close'].resample(RESAMPLE_RULE).last()
df['Volume']= raw['Volume'].resample(RESAMPLE_RULE).sum()
df.dropna(inplace=True)
# -----------------------------
# Indicators
# -----------------------------
df['EMA200'] = ema(df['Close'], EMA_TREND_LEN)
df['RSI'] = rsi(df['Close'], RSI_LEN)
# MACD (12,26,9)
ema12 = ema(df['Close'], 12)
ema26 = ema(df['Close'], 26)
df['MACD'] = ema12 - ema26
df['MACDsig'] = ema(df['MACD'], 9)
df['MACDh'] = df['MACD'] - df['MACDsig']
# ATR
df['ATR'] = atr(df, 14)
🔹 4. Signals + Backtest Loop
# -----------------------------
# Entry / Exit conditions (Long-only)
# -----------------------------
df['trend_up'] = df['Close'] > df['EMA200']
df['rsi_cross_up'] = (df['RSI'].shift(1) < 30) & (df['RSI'] >= 30)
df['macd_cross_up'] = (df['MACD'].shift(1) <= df['MACDsig'].shift(1)) & (df['MACD'] > df['MACDsig'])
df['exit_signal'] = (df['RSI'] > 70) | (df['MACDh'] < 0)
# -----------------------------
# Backtest loop
# -----------------------------
equity = INITIAL_EQUITY
position = 0
entry_price = np.nan
stop_price = np.nan
tp_price = np.nan
position_size = 0.0
trade_log = []
for ts, row in df.iterrows():
price = row['Close']
bar_atr = row['ATR']
if position == 1 and price - entry_price >= bar_atr:
stop_price = max(stop_price, entry_price)
if position == 1:
exit_reason = None
if price <= stop_price:
exit_reason = "stop"
elif price >= tp_price:
exit_reason = "tp"
elif row['exit_signal']:
exit_reason = "signal"
if exit_reason:
exit_exec_price = price * (1 - SLIPPAGE)
gross_pnl = (exit_exec_price - entry_price) * position_size
fees = (entry_price * position_size * FEE_RATE) + (exit_exec_price * position_size * FEE_RATE)
net_pnl = gross_pnl - fees
equity += net_pnl
trade_log.append({
"time": ts, "side": "SELL", "exit_reason": exit_reason,
"price": exit_exec_price, "size": position_size,
"gross_pnl": gross_pnl, "fees": fees, "net_pnl": net_pnl,
"equity": equity
})
position = 0
entry_price = np.nan
stop_price = np.nan
tp_price = np.nan
position_size = 0.0
continue
if position == 0 and row['trend_up'] and row['rsi_cross_up'] and row['macd_cross_up'] and bar_atr > 0:
stop_dist = ATR_MULT_STOP * bar_atr
tp_dist = ATR_MULT_TP * bar_atr
risk_amount = equity * RISK_PER_TRADE
size = risk_amount / stop_dist if stop_dist > 0 else 0
if size > 0:
entry_exec_price = price * (1 + SLIPPAGE)
fees = entry_exec_price * size * FEE_RATE
equity_after_fee = equity - fees
if equity_after_fee > 0:
position = 1
entry_price = entry_exec_price
position_size = size
stop_price = entry_price - stop_dist
tp_price = entry_price + tp_dist
equity = equity_after_fee
trade_log.append({
"time": ts, "side": "BUY", "price": entry_price,
"size": position_size, "fees": fees, "equity": equity
})
🔹 5. Performance + Report
# -----------------------------
# Performance metrics
# -----------------------------
trades = pd.DataFrame(trade_log)
wins = trades[trades.get("net_pnl", 0) > 0]
total_net = trades.get("net_pnl", pd.Series(dtype=float)).sum()
num_roundtrips = (trades['side'] == 'SELL').sum()
df['equity'] = np.nan
eq = INITIAL_EQUITY
for ts, row in df.iterrows():
realized = trades[(trades['time'] == ts) & (trades['side'] == 'SELL')]['net_pnl'].sum() if not trades.empty else 0.0
eq += realized
df.at[ts, 'equity'] = eq
def max_drawdown(series):
roll_max = series.cummax()
dd = series / roll_max - 1.0
return dd.min()
mdd = max_drawdown(df['equity'].dropna()) if df['equity'].notna().sum() > 0 else 0.0
bar_returns = df['equity'].pct_change().dropna()
if not bar_returns.empty:
mean_r = bar_returns.mean()
std_r = bar_returns.std()
bars_per_year = int(365*24/4)
sharpe = (mean_r / std_r) * math.sqrt(bars_per_year) if std_r > 0 else float('nan')
else:
sharpe = float('nan')
# -----------------------------
# Report
# -----------------------------
print("\n=== BTC Strategy Backtest Summary ({} / {}) ===".format(SYMBOL, RESAMPLE_RULE))
print(f"Number of bars : {len(df)}")
print(f"Number of roundtrips : {num_roundtrips}")
last_eq = df['equity'].dropna().iloc[-1] if df['equity'].notna().sum() > 0 else INITIAL_EQUITY
print(f"Final Equity : {last_eq:.2f} (starting {INITIAL_EQUITY:.2f})")
print(f"Total Net P&L : {total_net:.2f}")
if num_roundtrips > 0:
print(f"Avg P&L per Trade : {total_net/num_roundtrips:.2f}")
winrate = len(wins) / num_roundtrips * 100 if num_roundtrips > 0 else 0.0
print(f"Win Rate : {winrate:.1f}%")
print(f"Max Drawdown : {mdd*100:.2f}%")
print(f"Approx Sharpe : {sharpe:.2f}")
if not trades.empty:
trades.to_csv("btc_trades.csv", index=False)
print("\nTrade log saved: btc_trades.csv")