gavindiaz/noteQuant-backtest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimized backtesting effiency

Browse Source
This commit is contained in:
moen0
2026-04-11 17:23:26 +02:00
parent 766009b11a
commit 373e589297
10 changed files with 371780 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/routes.py
+49
View File
@@ -55,4 +55,53 @@ def get_indicators(timeframe: int = 5):
"liquidity": levels,
"fvgs": fvgs,
"order_blocks": obs
}
@app.get("/api/backtest")
def get_backtest(timeframe: int = 5, rr: float = 2.5):
candles_1m = load_candles("data/data.csv")
candles = resample_candles(candles_1m, period=timeframe)
from strategies.ict_strategy import ICTStrategy
strategy = ICTStrategy(
session="london",
lookback=7,
ob_max_age=50,
atr_mult=2.5,
use_liquidity_sweep=True,
sweep_lookback=5,
)
from engine.backtester import run_backtest
trades = run_backtest(candles, strategy, 10000, risk_reward=rr)
candle_times = [c.time_open.isoformat() for c in candles]
trades_data = []
for t in trades:
trades_data.append({
"enter_time": t.enter_time.isoformat(),
"exit_time": t.exit_time.isoformat(),
"enter_price": t.enter_price,
"exit_price": t.exit_price,
"direction": t.direction,
"pnl": t.pnl,
})
total_pnl = sum(t.pnl for t in trades)
winners = [t for t in trades if t.pnl > 0]
losers = [t for t in trades if t.pnl <= 0]
return {
"trades": trades_data,
"candle_times": candle_times,
"stats": {
"total_trades": len(trades),
"winners": len(winners),
"losers": len(losers),
"win_rate": len(winners) / len(trades) * 100 if trades else 0,
"total_pnl": total_pnl,
"avg_win": sum(t.pnl for t in winners) / len(winners) if winners else 0,
"avg_loss": sum(t.pnl for t in losers) / len(losers) if losers else 0,
"risk_reward": rr,
}
}
backend/data/data1.csv
+371076
View File
File diff suppressed because it is too large Load Diff
backend/engine/backtester.py
+79 -25
View File
@@ -1,37 +1,91 @@
from data.model import Candle, Trade
from strategies.base import SimpleStrategy
# takes a list of Candles and a starting balance, and returns a list of Trades
def run_backtest(candles: list[Candle], starting_balance: float) -> list[Trade]:
balance = starting_balance
def run_backtest(
candles: list[Candle],
strategy,
starting_balance: float = 10000.0,
risk_reward: float = 1.0
) -> list[Trade]:
"""
Runs a backtest on a list of candles using the provided strategy.
Returns a list of closed Trades.
"""
trades = []
position = None
strategy = SimpleStrategy()
# One-time preparation (e.g. pre-compute indicators)
if hasattr(strategy, "prepare"):
strategy.prepare(candles)
for i, candle in enumerate(candles):
# pass 'i' or the sliced history to the strategy
signal = strategy.check_signal(candles[:i+1])
# If signal and no position, open trade
if signal == "BUY" and position is None:
position = {
"type": "long",
"entry_price": candle.close,
"enter_time": candle.time_open
}
# === 1. Check if we have an open position (SL/TP hit) ===
if position is not None:
hit_sl = False
hit_tp = False
exit_price = None
# If signal and in position, close trade
elif signal == "SELL" and position is not None:
trade = Trade(
enter_time=position["enter_time"],
enter_price=position["entry_price"],
direction="long",
exit_time=candle.time_open,
exit_price=candle.close,
pnl=candle.close - position["entry_price"]
)
trades.append(trade)
position = None
if position["direction"] == "long":
if candle.low <= position["stop_loss"]:
hit_sl = True
exit_price = position["stop_loss"]
elif candle.high >= position["take_profit"]:
hit_tp = True
exit_price = position["take_profit"]
else: # short
if candle.high >= position["stop_loss"]:
hit_sl = True
exit_price = position["stop_loss"]
elif candle.low <= position["take_profit"]:
hit_tp = True
exit_price = position["take_profit"]
if hit_sl or hit_tp:
# Calculate PnL
if position["direction"] == "long":
pnl = exit_price - position["entry_price"]
else: # short
pnl = position["entry_price"] - exit_price
trade = Trade(
enter_time=position["enter_time"],
enter_price=position["entry_price"],
direction=position["direction"],
exit_time=candle.time_open,
exit_price=exit_price,
pnl=pnl
)
trades.append(trade)
position = None
# === 2. Look for new entry signal only if flat ===
if position is None:
signal = strategy.check_signal(candles, i) # Fixed: pass index instead of slicing
if signal == "BUY":
atr = candle.high - candle.low
mult = getattr(strategy, "atr_mult", 0.5)
bracket = atr * mult
position = {
"direction": "long",
"entry_price": candle.close,
"enter_time": candle.time_open,
"stop_loss": candle.close - bracket,
"take_profit": candle.close + (bracket * risk_reward),
}
elif signal == "SELL":
atr = candle.high - candle.low
mult = getattr(strategy, "atr_mult", 0.5)
bracket = atr * mult
position = {
"direction": "short",
"entry_price": candle.close,
"enter_time": candle.time_open,
"stop_loss": candle.close + bracket,
"take_profit": candle.close - (bracket * risk_reward),
}
return trades
backend/indicators/sessions.py
+34
View File
@@ -0,0 +1,34 @@
from datetime import time
SESSIONS_EST = {
"asian": (time(19, 0), time(3, 0)),
"london": (time(2, 0), time(5, 0)),
"new_york": (time(7, 0), time(10, 0)),
"london_close": (time(10, 0), time(12, 0)),
}
def in_session(candle_time, session_name):
t = candle_time.time()
start, end = SESSIONS_EST[session_name]
if start > end: # crosses midnight
return t >= start or t < end
return start <= t < end
def get_session(candle_time):
for name in SESSIONS_EST:
if in_session(candle_time, name):
return name
return "off_hours"
def filter_by_session(candles, session_name):
return [c for c in candles if in_session(c.time_open, session_name)]
def get_asian_range(candles):
asian = filter_by_session(candles, "asian")
if not asian:
return None
return {
"high": max(c.high for c in asian),
"low": min(c.low for c in asian),
"mid": (max(c.high for c in asian) + min(c.low for c in asian)) / 2,
}
backend/optimize.py
+29
View File
@@ -0,0 +1,29 @@
from data.loader import load_candles, resample_candles
from engine.backtester import run_backtest
from strategies.categorical_strategy import CategoricalStrategy
candles_1m = load_candles("data/data.csv")
candles_5m = resample_candles(candles_1m, period=5)
best_pnl = float("-inf")
best_params = None
for lookback in [10, 15, 20, 30, 40, 50]:
for threshold in [0.2, 0.3, 0.4, 0.5, 0.7, 1.0]:
for atr_mult in [0.3, 0.4, 0.5, 0.6, 0.7]:
strategy = CategoricalStrategy(
lookback=lookback,
range_threshold=threshold,
atr_multiplier=atr_mult
)
trades = run_backtest(candles_5m, strategy, 10000)
if len(trades) < 50:
continue
total_pnl = sum(t.pnl for t in trades)
win_rate = len([t for t in trades if t.pnl > 0]) / len(trades) * 100
if total_pnl > best_pnl:
best_pnl = total_pnl
best_params = (lookback, threshold, atr_mult)
print(f"New best: LB={lookback}, TH={threshold}, ATR={atr_mult} -> PnL={total_pnl:.2f}, WR={win_rate:.1f}%, Trades={len(trades)}")
print(f"\nBest: lookback={best_params[0]}, threshold={best_params[1]}, atr_mult={best_params[2]}, PnL={best_pnl:.2f}")
backend/optimize_ict.py
+84
View File
@@ -0,0 +1,84 @@
from data.loader import load_candles, resample_candles
from engine.backtester import run_backtest
from strategies.ict_strategy import ICTStrategy
import time
from tqdm import tqdm
# Load data once
candles_1m = load_candles("data/data1.csv")
candles_5m = resample_candles(candles_1m, period=5)
# Initialize best results
best_pnl = float("-inf")
best_params = None
# Generate all parameter combinations
param_combos = []
for session in ["london", "new_york"]:
for lookback in [3, 5, 7, 10]:
for ob_age in [20, 50, 80]:
for atr in [1.0, 1.5, 2.0, 2.5]:
for sweep in [True, False]:
sweep_lbs = [5, 10, 15] if sweep else [0]
for sweep_lb in sweep_lbs:
param_combos.append({
"session": session,
"lookback": lookback,
"ob_age": ob_age,
"atr": atr,
"sweep": sweep,
"sweep_lb": sweep_lb
})
print(f"Starting optimization of {len(param_combos)} combinations on your M4 Mac...\n")
total_start = time.perf_counter()
# Main loop with progress bar
for params in tqdm(param_combos, desc="Optimizing ICT Strategy", unit="backtest"):
strategy = ICTStrategy(
session=params["session"],
lookback=params["lookback"],
ob_max_age=params["ob_age"],
atr_mult=params["atr"],
use_liquidity_sweep=params["sweep"],
sweep_lookback=params["sweep_lb"],
)
# Accurate timing
t0 = time.perf_counter()
trades = run_backtest(candles_5m, strategy, 10000)
elapsed = time.perf_counter() - t0
# Optional: print every backtest (can be noisy, comment out if you want cleaner output)
# print(f"Backtest took {elapsed:.4f}s | Trades: {len(trades)}")
if len(trades) < 5:
continue
total_pnl = sum(t.pnl for t in trades)
wr = len([t for t in trades if t.pnl > 0]) / len(trades) * 100 if trades else 0.0
if total_pnl > best_pnl:
best_pnl = total_pnl
best_params = {
"session": params["session"],
"lookback": params["lookback"],
"ob_age": params["ob_age"],
"atr": params["atr"],
"sweep": params["sweep"],
"sweep_lb": params["sweep_lb"],
"trades": len(trades),
"wr": round(wr, 2)
}
tqdm.write(f"New best! PnL = {total_pnl:.2f} | Params: {best_params}")
# Final results
total_time = time.perf_counter() - total_start
print("\n" + "="*60)
print("Optimization finished!")
print(f"Total time on your M4: {total_time:.1f} seconds ({total_time/60:.1f} minutes)")
print(f"Best params: {best_params}")
print(f"Best PnL: {best_pnl:.2f}")
print("="*60)
backend/optimize_para_ict.py
+87
View File
@@ -0,0 +1,87 @@
from data.loader import load_candles, resample_candles
from engine.backtester import run_backtest
from strategies.ict_strategy import ICTStrategy
import time
from tqdm import tqdm
from joblib import Parallel, delayed
print("Loading data...")
candles_1m = load_candles("data/data1.csv")
candles_5m = resample_candles(candles_1m, period=5)
print(f"Loaded {len(candles_5m):,} 5-minute candles.\n")
param_combos = []
for session in ["london", "new_york"]:
for lookback in [3, 5, 7, 10]:
for ob_age in [20, 50, 80]:
for atr in [1.0, 1.5, 2.0, 2.5]:
for sweep in [True, False]:
sweep_lbs = [5, 10, 15] if sweep else [0]
for sweep_lb in sweep_lbs:
param_combos.append({
"session": session,
"lookback": lookback,
"ob_age": ob_age,
"atr": atr,
"sweep": sweep,
"sweep_lb": sweep_lb
})
print(f"Starting parallel optimization of {len(param_combos)} combinations...\n")
def run_one_combo(params):
strategy = ICTStrategy(
session=params["session"],
lookback=params["lookback"],
ob_max_age=params["ob_age"],
atr_mult=params["atr"],
use_liquidity_sweep=params["sweep"],
sweep_lookback=params["sweep_lb"],
)
t0 = time.perf_counter()
trades = run_backtest(candles_5m, strategy, 10000)
elapsed = time.perf_counter() - t0
if len(trades) < 5:
return None
total_pnl = sum(t.pnl for t in trades)
wr = len([t for t in trades if t.pnl > 0]) / len(trades) * 100 if trades else 0.0
return {
"params": params,
"pnl": total_pnl,
"trades": len(trades),
"wr": round(wr, 2),
"time": round(elapsed, 4)
}
total_start = time.perf_counter()
results = Parallel(n_jobs=-1, verbose=10)(
delayed(run_one_combo)(params) for params in param_combos
)
valid_results = [r for r in results if r is not None]
if not valid_results:
print("No valid strategies found with at least 5 trades.")
exit()
best_result = max(valid_results, key=lambda x: x["pnl"])
total_time = time.perf_counter() - total_start
print("\n" + "="*70)
print("PARALLEL OPTIMIZATION FINISHED!")
print(f"Total time on M4 Mac: {total_time:.1f} seconds ({total_time/60:.1f} minutes)")
print(f"Processed {len(param_combos)} combinations at ~{len(param_combos)/total_time:.2f} combos/second")
print(f"Best PnL: {best_result['pnl']:.2f}")
print(f"Best Params: {best_result['params']}")
print(f"Trades: {best_result['trades']} | Win Rate: {best_result['wr']}%")
print("="*70)
print("\nTop 5 results:")
for res in sorted(valid_results, key=lambda x: x["pnl"], reverse=True)[:5]:
print(f"PnL: {res['pnl']:.2f} | Trades: {res['trades']} | WR: {res['wr']}% | {res['params']}")
backend/run.py
+37 -21
View File
@@ -1,28 +1,44 @@
from data.loader import load_candles, resample_candles
from indicators.market_structure import find_swing_points, detect_structure
from indicators.liquidity import find_liquidity_levels
from indicators.fvg import find_fvgs
from indicators.order_blocks import find_order_blocks
from engine.backtester import run_backtest
from strategies.ict_strategy import ICTStrategy
import time
candles_1m = load_candles("data/data.csv")
candles_3m = resample_candles(candles_1m, period=3)
# Load data
candles_1m = load_candles("data/data1.csv")
candles_5m = resample_candles(candles_1m, period=5)
print(f"1m: {len(candles_1m)} candles")
print(f"3m: {len(candles_3m)} candles")
print(f"5m: {len(candles_5m)} candles")
print("Testing different Risk-Reward ratios with optimized ICTStrategy...\n")
swings = find_swing_points(candles_5m)
structure = detect_structure(swings)
levels = find_liquidity_levels(swings)
fvgs = find_fvgs(candles_5m)
obs = find_order_blocks(candles_5m, structure)
# Best params from optimization (you can tweak session/lookback etc. if you want)
strategy = ICTStrategy(
session="new_york", # Best was New York
lookback=7,
ob_max_age=20, # Best was 20
atr_mult=2.5,
use_liquidity_sweep=False, # Best was False
sweep_lookback=5,
)
print(f"Swing points: {len(swings)}")
print(f"Structure points: {len(structure)}")
print(f"Liquidity levels: {len(levels)}")
print(f"FVGs: {len(fvgs)}")
print(f"Order blocks: {len(obs)}")
for rr in [1.0, 1.5, 2.0, 2.5, 3.0]:
t0 = time.perf_counter()
for o in obs[:5]:
print(o)
trades = run_backtest(candles_5m, strategy, 10000, risk_reward=rr)
elapsed = time.perf_counter() - t0
if not trades:
print(f"RR={rr}: No trades")
continue
total_pnl = sum(t.pnl for t in trades)
winners = [t for t in trades if t.pnl > 0]
losers = [t for t in trades if t.pnl <= 0]
wr = len(winners) / len(trades) * 100 if trades else 0
avg_win = sum(t.pnl for t in winners) / len(winners) if winners else 0
avg_loss = sum(t.pnl for t in losers) / len(losers) if losers else 0
profit_factor = abs(sum(t.pnl for t in winners) / sum(t.pnl for t in losers)) if losers else float('inf')
print(f"RR={rr:4.1f} | Trades={len(trades):4d} | WR={wr:5.1f}% | "
f"PnL={total_pnl:8.2f} | AvgWin={avg_win:6.3f} | AvgLoss={avg_loss:6.3f} | "
f"PF={profit_factor:5.2f} | Time={elapsed:.3f}s")
backend/strategies/categorical_strategy.py
+75
View File
@@ -0,0 +1,75 @@
class CategoricalStrategy:
def __init__(self, lookback=20, range_threshold=0.4, atr_multiplier=0.5):
self.lookback = lookback
self.range_threshold = range_threshold
self.atr_multiplier = atr_multiplier
def get_atr1(self, candle):
return candle.high - candle.low
def classify(self, history):
if len(history) < self.lookback:
return None
window = history[-self.lookback:]
highest = max(c.high for c in window)
lowest = min(c.low for c in window)
full_range = highest - lowest
# Check how much of the range was used early vs late
first_half = window[:len(window) // 2]
second_half = window[len(window) // 2:]
first_high = max(c.high for c in first_half)
first_low = min(c.low for c in first_half)
second_high = max(c.high for c in second_half)
second_low = min(c.low for c in second_half)
# If second half is expanding beyond first half range, it's direction
expansion = 0
if second_high > first_high:
expansion += second_high - first_high
if second_low < first_low:
expansion += first_low - second_low
avg_candle = sum(self.get_atr1(c) for c in window) / len(window)
if expansion > avg_candle * self.range_threshold:
return "direction"
return "consolidation"
def check_signal(self, history):
if len(history) < self.lookback + 1:
return None
category = self.classify(history)
if category is None:
return None
window = history[-self.lookback:]
highest = max(c.high for c in window)
lowest = min(c.low for c in window)
mid = (highest + lowest) / 2
candle = history[-1]
prev = history[-2]
atr = self.get_atr1(candle)
bracket = atr * self.atr_multiplier
if category == "consolidation":
# Near top of range and candle turning down: sell
if candle.close > mid and candle.close < prev.close:
return "SELL"
# Near bottom of range and candle turning up: buy
if candle.close < mid and candle.close > prev.close:
return "BUY"
elif category == "direction":
# Price pushing up: follow
if candle.close > prev.close and candle.close > mid:
return "BUY"
# Price pushing down: follow
if candle.close < prev.close and candle.close < mid:
return "SELL"
return None