reorganized data files and enhance backtesting structure, monte carlo sim
This commit is contained in:
moen0
+974
-61
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
|
Can't render this file because it is too large.
|
File diff suppressed because it is too large
Load Diff
@@ -5,6 +5,7 @@ from data.model import Candle
|
||||
def load_candles(filepath: str) -> list[Candle]:
|
||||
df = pd.read_csv(filepath, sep=";", header=None, names=["timestamp", "open", "high", "low", "close", "volume"])
|
||||
df["timestamp"] = pd.to_datetime(df["timestamp"], format="%Y%m%d %H%M%S")
|
||||
df = df.sort_values("timestamp", kind="mergesort").drop_duplicates(subset=["timestamp"], keep="last")
|
||||
|
||||
candles = []
|
||||
for _, row in df.iterrows():
|
||||
|
||||
+14
-2
@@ -1,20 +1,32 @@
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
@dataclass
|
||||
class Candle:
|
||||
time_open: datetime
|
||||
open: float
|
||||
high: float
|
||||
low: float
|
||||
close: datetime
|
||||
close: float
|
||||
volume: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class Signal:
|
||||
direction: str
|
||||
stop_loss: float
|
||||
entry_price: float
|
||||
|
||||
|
||||
@dataclass
|
||||
class Trade:
|
||||
enter_time: datetime
|
||||
enter_price: float
|
||||
direction: str
|
||||
exit_time:datetime
|
||||
exit_time: datetime
|
||||
exit_price: float
|
||||
pnl: float
|
||||
r_multiple: float = 0.0
|
||||
partial_tp_taken: bool = False
|
||||
partial_tp_realized_pnl: float = 0.0
|
||||
|
||||
+265
-64
@@ -1,51 +1,222 @@
|
||||
from data.model import Candle, Trade
|
||||
from strategies.base import SimpleStrategy
|
||||
from data.model import Trade
|
||||
from collections import defaultdict
|
||||
import math
|
||||
|
||||
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.
|
||||
"""
|
||||
|
||||
def _apply_break_even_if_triggered(position, candle, strategy):
|
||||
if not position:
|
||||
return
|
||||
|
||||
if not getattr(strategy, "use_break_even", False):
|
||||
return
|
||||
|
||||
if position.get("break_even_armed"):
|
||||
return
|
||||
|
||||
trigger_rr = float(getattr(strategy, "be_trigger_rr", 1.0) or 0.0)
|
||||
if trigger_rr <= 0:
|
||||
return
|
||||
|
||||
is_long = position["direction"] == "long"
|
||||
entry = position["entry_price"]
|
||||
risk_distance = max(position.get("risk_distance", 0.0), 0.0)
|
||||
if risk_distance <= 0:
|
||||
return
|
||||
|
||||
trigger_price = entry + (risk_distance * trigger_rr) if is_long else entry - (risk_distance * trigger_rr)
|
||||
reached_trigger = candle.high >= trigger_price if is_long else candle.low <= trigger_price
|
||||
if reached_trigger:
|
||||
position["stop_loss"] = entry
|
||||
position["break_even_armed"] = True
|
||||
|
||||
|
||||
def _apply_partial_tp_if_triggered(position, candle, strategy):
|
||||
if not position:
|
||||
return
|
||||
|
||||
if not getattr(strategy, "use_partial_tp", False):
|
||||
return
|
||||
|
||||
if position.get("partial_taken"):
|
||||
return
|
||||
|
||||
trigger_rr = float(getattr(strategy, "partial_tp_rr", 1.0) or 0.0)
|
||||
if trigger_rr <= 0:
|
||||
return
|
||||
|
||||
partial_pct = float(getattr(strategy, "partial_tp_percent", 0.0) or 0.0)
|
||||
if partial_pct <= 0:
|
||||
return
|
||||
|
||||
close_fraction = min(max(partial_pct / 100.0, 0.0), 1.0)
|
||||
remaining_fraction = max(position.get("remaining_fraction", 1.0), 0.0)
|
||||
if remaining_fraction <= 0:
|
||||
position["partial_taken"] = True
|
||||
return
|
||||
|
||||
close_fraction = min(close_fraction, remaining_fraction)
|
||||
if close_fraction <= 0:
|
||||
return
|
||||
|
||||
is_long = position["direction"] == "long"
|
||||
entry = position["entry_price"]
|
||||
risk_distance = max(position.get("risk_distance", 0.0), 0.0)
|
||||
if risk_distance <= 0:
|
||||
return
|
||||
|
||||
trigger_price = entry + (risk_distance * trigger_rr) if is_long else entry - (risk_distance * trigger_rr)
|
||||
reached_trigger = candle.high >= trigger_price if is_long else candle.low <= trigger_price
|
||||
if not reached_trigger:
|
||||
return
|
||||
|
||||
lot_size = max(position.get("lot_size", 0.0), 0.0)
|
||||
price_move = (trigger_price - entry) if is_long else (entry - trigger_price)
|
||||
realized_piece = price_move * lot_size * close_fraction
|
||||
|
||||
position["realized_pnl"] = position.get("realized_pnl", 0.0) + realized_piece
|
||||
position["remaining_fraction"] = max(0.0, remaining_fraction - close_fraction)
|
||||
position["partial_taken"] = True
|
||||
|
||||
|
||||
def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
|
||||
max_daily_loss=0.0, max_consecutive_losses=0, risk_pct=1.0):
|
||||
trades = []
|
||||
position = None
|
||||
|
||||
# One-time preparation (e.g. pre-compute indicators)
|
||||
consecutive_losses = 0
|
||||
daily_pnl = defaultdict(float)
|
||||
if hasattr(strategy, "prepare"):
|
||||
strategy.prepare(candles)
|
||||
|
||||
for i, candle in enumerate(candles):
|
||||
# === 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 position:
|
||||
_apply_partial_tp_if_triggered(position, candle, strategy)
|
||||
_apply_break_even_if_triggered(position, candle, strategy)
|
||||
|
||||
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"]
|
||||
is_long = position["direction"] == "long"
|
||||
sl, tp = position["stop_loss"], position["take_profit"]
|
||||
|
||||
hit_sl = candle.low <= sl if is_long else candle.high >= sl
|
||||
hit_tp = candle.high >= tp if is_long else candle.low <= tp
|
||||
|
||||
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
|
||||
exit_price = sl if hit_sl else tp
|
||||
price_move = (exit_price - position["entry_price"]) if is_long else (position["entry_price"] - exit_price)
|
||||
lot_size = max(position.get("lot_size", 0.0), 0.0)
|
||||
remaining_fraction = max(position.get("remaining_fraction", 1.0), 0.0)
|
||||
remaining_pnl = price_move * lot_size * remaining_fraction
|
||||
pnl = position.get("realized_pnl", 0.0) + remaining_pnl
|
||||
initial_risk = max(position.get("initial_risk_amount", 0.0), 1e-12)
|
||||
r_multiple = pnl / initial_risk
|
||||
|
||||
trades.append(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,
|
||||
r_multiple=r_multiple,
|
||||
partial_tp_taken=bool(position.get("partial_taken", False)),
|
||||
partial_tp_realized_pnl=float(position.get("realized_pnl", 0.0) or 0.0),
|
||||
))
|
||||
position = None
|
||||
|
||||
if pnl <= 0:
|
||||
consecutive_losses += 1
|
||||
else:
|
||||
consecutive_losses = 0
|
||||
|
||||
daily_pnl[candle.time_open.date()] += pnl
|
||||
|
||||
if position is None:
|
||||
if max_consecutive_losses > 0 and consecutive_losses >= max_consecutive_losses:
|
||||
continue
|
||||
if max_daily_loss > 0:
|
||||
loss_limit = starting_balance * (max_daily_loss / 100)
|
||||
if daily_pnl[candle.time_open.date()] <= -loss_limit:
|
||||
continue
|
||||
|
||||
signal = strategy.check_signal(candles, i)
|
||||
|
||||
if signal is not None:
|
||||
is_long = signal.direction == "BUY"
|
||||
entry = signal.entry_price
|
||||
sl = signal.stop_loss
|
||||
sl_distance = abs(entry - sl)
|
||||
if (
|
||||
sl_distance <= 0
|
||||
or not math.isfinite(sl_distance)
|
||||
or not math.isfinite(entry)
|
||||
or not math.isfinite(sl)
|
||||
or risk_pct <= 0
|
||||
):
|
||||
continue
|
||||
|
||||
risk_amount = starting_balance * (risk_pct / 100)
|
||||
if risk_amount <= 0 or not math.isfinite(risk_amount):
|
||||
continue
|
||||
|
||||
lot_size = risk_amount / sl_distance
|
||||
if lot_size <= 0 or not math.isfinite(lot_size):
|
||||
continue
|
||||
|
||||
tp = entry + (sl_distance * risk_reward) if is_long else entry - (sl_distance * risk_reward)
|
||||
|
||||
position = {
|
||||
"direction": "long" if is_long else "short",
|
||||
"entry_price": entry,
|
||||
"enter_time": candle.time_open,
|
||||
"stop_loss": sl,
|
||||
"take_profit": tp,
|
||||
"risk_distance": sl_distance,
|
||||
"lot_size": lot_size,
|
||||
"break_even_armed": False,
|
||||
"partial_taken": False,
|
||||
"remaining_fraction": 1.0,
|
||||
"realized_pnl": 0.0,
|
||||
"initial_risk_amount": risk_amount,
|
||||
}
|
||||
|
||||
return trades
|
||||
|
||||
|
||||
def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
|
||||
max_daily_loss=0.0, max_consecutive_losses=0, risk_pct=1.0):
|
||||
position = None
|
||||
consecutive_losses = 0
|
||||
daily_pnl = defaultdict(float)
|
||||
total = len(candles)
|
||||
|
||||
if hasattr(strategy, "prepare"):
|
||||
strategy.prepare(candles)
|
||||
|
||||
yield {"type": "start", "total_candles": total}
|
||||
|
||||
progress_interval = max(1, total // 50)
|
||||
|
||||
for i, candle in enumerate(candles):
|
||||
if i % progress_interval == 0:
|
||||
yield {"type": "progress", "processed_candles": i, "total_candles": total}
|
||||
|
||||
if position:
|
||||
_apply_partial_tp_if_triggered(position, candle, strategy)
|
||||
_apply_break_even_if_triggered(position, candle, strategy)
|
||||
|
||||
is_long = position["direction"] == "long"
|
||||
sl, tp = position["stop_loss"], position["take_profit"]
|
||||
|
||||
hit_sl = candle.low <= sl if is_long else candle.high >= sl
|
||||
hit_tp = candle.high >= tp if is_long else candle.low <= tp
|
||||
|
||||
if hit_sl or hit_tp:
|
||||
exit_price = sl if hit_sl else tp
|
||||
price_move = (exit_price - position["entry_price"]) if is_long else (position["entry_price"] - exit_price)
|
||||
lot_size = max(position.get("lot_size", 0.0), 0.0)
|
||||
remaining_fraction = max(position.get("remaining_fraction", 1.0), 0.0)
|
||||
remaining_pnl = price_move * lot_size * remaining_fraction
|
||||
pnl = position.get("realized_pnl", 0.0) + remaining_pnl
|
||||
initial_risk = max(position.get("initial_risk_amount", 0.0), 1e-12)
|
||||
r_multiple = pnl / initial_risk
|
||||
|
||||
trade = Trade(
|
||||
enter_time=position["enter_time"],
|
||||
@@ -53,39 +224,69 @@ def run_backtest(
|
||||
direction=position["direction"],
|
||||
exit_time=candle.time_open,
|
||||
exit_price=exit_price,
|
||||
pnl=pnl
|
||||
pnl=pnl,
|
||||
r_multiple=r_multiple,
|
||||
partial_tp_taken=bool(position.get("partial_taken", False)),
|
||||
partial_tp_realized_pnl=float(position.get("realized_pnl", 0.0) or 0.0),
|
||||
)
|
||||
trades.append(trade)
|
||||
position = None
|
||||
|
||||
# === 2. Look for new entry signal only if flat ===
|
||||
if pnl <= 0:
|
||||
consecutive_losses += 1
|
||||
else:
|
||||
consecutive_losses = 0
|
||||
|
||||
daily_pnl[candle.time_open.date()] += pnl
|
||||
|
||||
yield {"type": "trade", "trade": trade, "processed_candles": i, "total_candles": total}
|
||||
|
||||
if position is None:
|
||||
signal = strategy.check_signal(candles, i) # Fixed: pass index instead of slicing
|
||||
if max_consecutive_losses > 0 and consecutive_losses >= max_consecutive_losses:
|
||||
continue
|
||||
if max_daily_loss > 0:
|
||||
loss_limit = starting_balance * (max_daily_loss / 100)
|
||||
if daily_pnl[candle.time_open.date()] <= -loss_limit:
|
||||
continue
|
||||
|
||||
if signal == "BUY":
|
||||
atr = candle.high - candle.low
|
||||
mult = getattr(strategy, "atr_mult", 0.5)
|
||||
bracket = atr * mult
|
||||
signal = strategy.check_signal(candles, i)
|
||||
|
||||
if signal is not None:
|
||||
is_long = signal.direction == "BUY"
|
||||
entry = signal.entry_price
|
||||
sl = signal.stop_loss
|
||||
sl_distance = abs(entry - sl)
|
||||
if (
|
||||
sl_distance <= 0
|
||||
or not math.isfinite(sl_distance)
|
||||
or not math.isfinite(entry)
|
||||
or not math.isfinite(sl)
|
||||
or risk_pct <= 0
|
||||
):
|
||||
continue
|
||||
|
||||
risk_amount = starting_balance * (risk_pct / 100)
|
||||
if risk_amount <= 0 or not math.isfinite(risk_amount):
|
||||
continue
|
||||
|
||||
lot_size = risk_amount / sl_distance
|
||||
if lot_size <= 0 or not math.isfinite(lot_size):
|
||||
continue
|
||||
|
||||
tp = entry + (sl_distance * risk_reward) if is_long else entry - (sl_distance * risk_reward)
|
||||
|
||||
position = {
|
||||
"direction": "long",
|
||||
"entry_price": candle.close,
|
||||
"direction": "long" if is_long else "short",
|
||||
"entry_price": entry,
|
||||
"enter_time": candle.time_open,
|
||||
"stop_loss": candle.close - bracket,
|
||||
"take_profit": candle.close + (bracket * risk_reward),
|
||||
"stop_loss": sl,
|
||||
"take_profit": tp,
|
||||
"risk_distance": sl_distance,
|
||||
"lot_size": lot_size,
|
||||
"break_even_armed": False,
|
||||
"partial_taken": False,
|
||||
"remaining_fraction": 1.0,
|
||||
"realized_pnl": 0.0,
|
||||
"initial_risk_amount": risk_amount,
|
||||
}
|
||||
|
||||
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
|
||||
yield {"type": "done", "total_candles": total}
|
||||
@@ -0,0 +1,154 @@
|
||||
import random
|
||||
|
||||
def _calculate_percentile(data, percentile):
|
||||
if not data: return 0.0
|
||||
sorted_data = sorted(data)
|
||||
index = (len(sorted_data) - 1) * (percentile / 100.0)
|
||||
lower = int(index)
|
||||
upper = lower + 1
|
||||
if upper >= len(sorted_data): return sorted_data[-1]
|
||||
return sorted_data[lower] + (index - lower) * (sorted_data[upper] - sorted_data[lower])
|
||||
|
||||
def _run_metrics(pnls, starting_balance):
|
||||
if not pnls:
|
||||
return {"net_pnl": 0.0, "win_rate": 0.0, "profit_factor": 0.0, "max_drawdown_pct": 0.0}
|
||||
|
||||
winners = [p for p in pnls if p > 0]
|
||||
losers = [p for p in pnls if p < 0]
|
||||
|
||||
net_pnl = sum(pnls)
|
||||
trade_count = len(pnls)
|
||||
win_rate = (len(winners) / trade_count) * 100 if trade_count > 0 else 0.0
|
||||
|
||||
gross_profit = sum(winners)
|
||||
gross_loss = abs(sum(losers))
|
||||
profit_factor = (gross_profit / gross_loss) if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0)
|
||||
|
||||
equity = starting_balance
|
||||
peak = starting_balance
|
||||
max_drawdown_pct = 0.0
|
||||
|
||||
for pnl in pnls:
|
||||
equity += pnl
|
||||
if equity > peak:
|
||||
peak = equity
|
||||
drawdown_pct = ((peak - equity) / peak) * 100 if peak > 0 else 0.0
|
||||
if drawdown_pct > max_drawdown_pct:
|
||||
max_drawdown_pct = drawdown_pct
|
||||
|
||||
return {
|
||||
"net_pnl": net_pnl,
|
||||
"win_rate": win_rate,
|
||||
"profit_factor": profit_factor,
|
||||
"max_drawdown_pct": max_drawdown_pct,
|
||||
}
|
||||
|
||||
def run_monte_carlo(
|
||||
trade_r_multiples,
|
||||
runs=1000,
|
||||
starting_balance=10000.0,
|
||||
risk_per_trade_pct=1.0,
|
||||
sampling_method="bootstrap",
|
||||
missed_trade_pct=5.0,
|
||||
pnl_variation_pct=10.0,
|
||||
price_noise_pct=0.0,
|
||||
slippage_per_trade=0.0,
|
||||
spread_per_trade=0.0,
|
||||
per_trade_cost=None,
|
||||
ruin_drawdown_pct=20.0,
|
||||
seed=None,
|
||||
):
|
||||
if not trade_r_multiples:
|
||||
return {"summary": {"runs": 0}, "distribution": [], "sample_runs": []}
|
||||
|
||||
run_count = max(1, int(runs))
|
||||
risk_pct = max(0.0, float(risk_per_trade_pct)) / 100.0
|
||||
pnl_var = max(0.0, float(pnl_variation_pct)) / 100.0
|
||||
price_var = max(0.0, float(price_noise_pct)) / 100.0
|
||||
miss_pct = max(0.0, float(missed_trade_pct)) / 100.0
|
||||
ruin_threshold = max(0.0, float(ruin_drawdown_pct))
|
||||
fixed_cost = float(per_trade_cost) if per_trade_cost is not None else (max(0.0, float(slippage_per_trade)) + max(0.0, float(spread_per_trade)))
|
||||
effective_var = max(pnl_var, price_var)
|
||||
|
||||
rng = random.Random(seed)
|
||||
run_results = []
|
||||
base_trades = list(trade_r_multiples)
|
||||
trade_count = len(base_trades)
|
||||
|
||||
for run_idx in range(run_count):
|
||||
# 1. Generate the Trade Sequence
|
||||
if sampling_method == "bootstrap":
|
||||
path = [rng.choice(base_trades) for _ in range(trade_count)]
|
||||
elif sampling_method == "shuffle":
|
||||
path = base_trades[:]
|
||||
rng.shuffle(path)
|
||||
else:
|
||||
path = base_trades[:]
|
||||
|
||||
adjusted_pnls = []
|
||||
equity = float(starting_balance)
|
||||
peak = float(starting_balance)
|
||||
ruin_hit = False
|
||||
|
||||
# 2. Execute the trades sequentially
|
||||
for base_r in path:
|
||||
# Execution Risk: Did the broker drop our connection?
|
||||
if rng.random() < miss_pct:
|
||||
continue
|
||||
|
||||
r_multiple = float(base_r)
|
||||
|
||||
# Add volatility noise to the outcome (Slippage)
|
||||
if effective_var > 0:
|
||||
r_multiple *= rng.uniform(1 - effective_var, 1 + effective_var)
|
||||
|
||||
# Calculate PnL in dollars based on CURRENT equity (Compounding)
|
||||
pnl = (equity * risk_pct * r_multiple) - fixed_cost
|
||||
|
||||
adjusted_pnls.append(pnl)
|
||||
equity += pnl
|
||||
|
||||
# 3. Live Drawdown & Ruin Check (Prevents Zombie Trading)
|
||||
if equity > peak:
|
||||
peak = equity
|
||||
|
||||
current_dd = ((peak - equity) / peak) * 100 if peak > 0 else 0.0
|
||||
|
||||
if current_dd >= ruin_threshold or equity <= 0:
|
||||
ruin_hit = True
|
||||
break # Account blown or max DD hit. STOP trading.
|
||||
|
||||
# Calculate metrics for the surviving trades
|
||||
metrics = _run_metrics(adjusted_pnls, starting_balance)
|
||||
metrics["run"] = run_idx + 1
|
||||
metrics["ruin"] = ruin_hit
|
||||
metrics["trades_taken"] = len(adjusted_pnls)
|
||||
run_results.append(metrics)
|
||||
|
||||
# --- Aggregate Statistics ---
|
||||
pnls = [r["net_pnl"] for r in run_results]
|
||||
dds = [r["max_drawdown_pct"] for r in run_results]
|
||||
|
||||
profitable_runs = sum(1 for p in pnls if p > 0)
|
||||
ruin_count = sum(1 for r in run_results if r["ruin"])
|
||||
|
||||
summary = {
|
||||
"runs": run_count,
|
||||
"avg_pnl": round(sum(pnls) / run_count, 2),
|
||||
"worst_case_pnl_5th_pct": round(_calculate_percentile(pnls, 5), 2), # 95% Confidence you make at least this much
|
||||
"profitable_run_pct": round((profitable_runs / run_count) * 100, 2),
|
||||
|
||||
"avg_max_drawdown_pct": round(sum(dds) / run_count, 2),
|
||||
"worst_case_dd_95th_pct": round(_calculate_percentile(dds, 95), 2), # 95% Confidence your DD won't exceed this
|
||||
"worst_max_drawdown_pct": round(max(dds), 2),
|
||||
|
||||
"avg_win_rate": round(sum(r["win_rate"] for r in run_results) / run_count, 2),
|
||||
"avg_profit_factor": round(sum(r["profit_factor"] for r in run_results) / run_count, 2),
|
||||
"probability_of_ruin": round((ruin_count / run_count) * 100, 2),
|
||||
}
|
||||
|
||||
return {
|
||||
"summary": summary,
|
||||
"distribution": run_results,
|
||||
"sample_runs": run_results,
|
||||
}
|
||||
+54
-16
@@ -1,27 +1,65 @@
|
||||
def find_fvgs(candles):
|
||||
def find_fvgs(candles, min_gap_size=0.0, impulse_multiplier=0.0):
|
||||
"""
|
||||
Find Fair Value Gaps in candle data.
|
||||
|
||||
Args:
|
||||
candles: list of Candle objects
|
||||
min_gap_size: minimum gap size in price units to filter noise (0 = no filter)
|
||||
impulse_multiplier: minimum body-to-avg ratio for the middle candle (0 = no filter)
|
||||
"""
|
||||
fvgs = []
|
||||
|
||||
avg_body = 0
|
||||
if impulse_multiplier > 0 and len(candles) > 20:
|
||||
bodies = [abs(c.close - c.open) for c in candles[:20]]
|
||||
avg_body = sum(bodies) / len(bodies)
|
||||
|
||||
for i in range(2, len(candles)):
|
||||
c1 = candles[i - 2]
|
||||
c2 = candles[i - 1]
|
||||
c3 = candles[i]
|
||||
|
||||
# Bullish
|
||||
# Impulse check on middle candle
|
||||
if impulse_multiplier > 0 and avg_body > 0:
|
||||
middle_body = abs(c2.close - c2.open)
|
||||
if middle_body < avg_body * impulse_multiplier:
|
||||
continue
|
||||
# Update rolling average
|
||||
avg_body = (avg_body * 19 + middle_body) / 20
|
||||
|
||||
# Bullish FVG
|
||||
if c1.high < c3.low:
|
||||
fvgs.append({
|
||||
"index": i - 1,
|
||||
"type": "bullish",
|
||||
"top": c3.low,
|
||||
"bottom": c1.high
|
||||
})
|
||||
gap_size = c3.low - c1.high
|
||||
if gap_size >= min_gap_size:
|
||||
fvgs.append({
|
||||
"index": i - 1,
|
||||
"type": "bullish",
|
||||
"top": c3.low,
|
||||
"bottom": c1.high,
|
||||
"mitigated": False,
|
||||
})
|
||||
|
||||
# bearish
|
||||
# Bearish FVG
|
||||
elif c1.low > c3.high:
|
||||
fvgs.append({
|
||||
"index": i - 1,
|
||||
"type": "bearish",
|
||||
"top": c1.low,
|
||||
"bottom": c3.high
|
||||
})
|
||||
gap_size = c1.low - c3.high
|
||||
if gap_size >= min_gap_size:
|
||||
fvgs.append({
|
||||
"index": i - 1,
|
||||
"type": "bearish",
|
||||
"top": c1.low,
|
||||
"bottom": c3.high,
|
||||
"mitigated": False,
|
||||
})
|
||||
|
||||
return fvgs
|
||||
# Mark mitigated FVGs
|
||||
for fvg in fvgs:
|
||||
if fvg["mitigated"]:
|
||||
continue
|
||||
if fvg["type"] == "bullish":
|
||||
if c3.low <= fvg["bottom"]:
|
||||
fvg["mitigated"] = True
|
||||
elif fvg["type"] == "bearish":
|
||||
if c3.high >= fvg["top"]:
|
||||
fvg["mitigated"] = True
|
||||
|
||||
return fvgs
|
||||
|
||||
@@ -19,7 +19,7 @@ def find_liquidity_levels(swings, tolerance=0.015, max_distance=100):
|
||||
"price": avg_price,
|
||||
"type": "equal_highs",
|
||||
"count": len(cluster),
|
||||
"indexes": [s["index"] for s in cluster]
|
||||
"indexes": [s["index"] for s in cluster],
|
||||
})
|
||||
used.add(i)
|
||||
|
||||
@@ -30,7 +30,7 @@ def find_liquidity_levels(swings, tolerance=0.015, max_distance=100):
|
||||
cluster = [l1]
|
||||
for j, l2 in enumerate(lows):
|
||||
if j != i and j not in used:
|
||||
if abs(h1["price"] - h2["price"]) <= tolerance and abs(h1["index"] - h2["index"]) <= max_distance:
|
||||
if abs(l1["price"] - l2["price"]) <= tolerance and abs(l1["index"] - l2["index"]) <= max_distance:
|
||||
cluster.append(l2)
|
||||
used.add(j)
|
||||
if len(cluster) >= 2:
|
||||
@@ -39,8 +39,8 @@ def find_liquidity_levels(swings, tolerance=0.015, max_distance=100):
|
||||
"price": avg_price,
|
||||
"type": "equal_lows",
|
||||
"count": len(cluster),
|
||||
"indexes": [s["index"] for s in cluster]
|
||||
"indexes": [s["index"] for s in cluster],
|
||||
})
|
||||
used.add(i)
|
||||
|
||||
return levels
|
||||
return levels
|
||||
|
||||
@@ -1,31 +1,42 @@
|
||||
def find_order_blocks(candles, structure, min_impulse=0.10):
|
||||
def find_order_blocks(candles, structure, min_impulse=0.10, min_ob_size=0.0):
|
||||
"""
|
||||
Find Order Blocks based on structure breaks.
|
||||
|
||||
Args:
|
||||
candles: list of Candle objects
|
||||
structure: list of structure points from detect_structure
|
||||
min_impulse: legacy param (unused, kept for compat)
|
||||
min_ob_size: minimum OB size in price units (0 = no filter)
|
||||
"""
|
||||
obs = []
|
||||
|
||||
for point in structure:
|
||||
if point["label"] == "HH":
|
||||
# Bullish break of structure, look back for last bearish candle
|
||||
idx = point["index"]
|
||||
for j in range(idx - 1, max(idx - 20, 0), -1):
|
||||
if candles[j].close < candles[j].open:
|
||||
obs.append({
|
||||
"index": j,
|
||||
"type": "bullish",
|
||||
"top": candles[j].open,
|
||||
"bottom": candles[j].close
|
||||
})
|
||||
size = candles[j].open - candles[j].close
|
||||
if size >= min_ob_size:
|
||||
obs.append({
|
||||
"index": j,
|
||||
"type": "bullish",
|
||||
"top": candles[j].open,
|
||||
"bottom": candles[j].close,
|
||||
})
|
||||
break
|
||||
|
||||
elif point["label"] == "LL":
|
||||
# Bearish break of structure, look back for last bullish candle
|
||||
idx = point["index"]
|
||||
for j in range(idx - 1, max(idx - 20, 0), -1):
|
||||
if candles[j].close > candles[j].open:
|
||||
obs.append({
|
||||
"index": j,
|
||||
"type": "bearish",
|
||||
"top": candles[j].close,
|
||||
"bottom": candles[j].open
|
||||
})
|
||||
size = candles[j].close - candles[j].open
|
||||
if size >= min_ob_size:
|
||||
obs.append({
|
||||
"index": j,
|
||||
"type": "bearish",
|
||||
"top": candles[j].close,
|
||||
"bottom": candles[j].open,
|
||||
})
|
||||
break
|
||||
|
||||
return obs
|
||||
return obs
|
||||
|
||||
@@ -5,24 +5,43 @@ SESSIONS_EST = {
|
||||
"london": (time(2, 0), time(5, 0)),
|
||||
"new_york": (time(7, 0), time(10, 0)),
|
||||
"london_close": (time(10, 0), time(12, 0)),
|
||||
"london_ny_overlap": (time(8, 0), time(10, 0)),
|
||||
}
|
||||
|
||||
|
||||
def in_session(candle_time, session_name):
|
||||
if session_name == "all":
|
||||
return True
|
||||
|
||||
if session_name not in SESSIONS_EST:
|
||||
return True
|
||||
|
||||
t = candle_time.time()
|
||||
start, end = SESSIONS_EST[session_name]
|
||||
if start > end: # crosses midnight
|
||||
if start > end:
|
||||
return t >= start or t < end
|
||||
return start <= t < end
|
||||
|
||||
|
||||
def get_session(candle_time):
|
||||
for name in SESSIONS_EST:
|
||||
if name == "all":
|
||||
continue
|
||||
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 in_day_filter(candle_time, allowed_days):
|
||||
if not allowed_days:
|
||||
return True
|
||||
return candle_time.weekday() in allowed_days
|
||||
|
||||
|
||||
def get_asian_range(candles):
|
||||
asian = filter_by_session(candles, "asian")
|
||||
if not asian:
|
||||
@@ -31,4 +50,4 @@ def get_asian_range(candles):
|
||||
"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,
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -2,7 +2,7 @@ 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_1m = load_candles("data/gbpjpy_jan.csv")
|
||||
candles_5m = resample_candles(candles_1m, period=5)
|
||||
|
||||
best_pnl = float("-inf")
|
||||
|
||||
@@ -45,14 +45,10 @@ for params in tqdm(param_combos, desc="Optimizing ICT Strategy", unit="backtest"
|
||||
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
|
||||
|
||||
|
||||
+15
-19
@@ -3,42 +3,38 @@ from engine.backtester import run_backtest
|
||||
from strategies.ict_strategy import ICTStrategy
|
||||
import time
|
||||
|
||||
# Load data
|
||||
candles_1m = load_candles("data/data1.csv")
|
||||
candles_1m = load_candles("data/2023gj.csv")
|
||||
candles_5m = resample_candles(candles_1m, period=5)
|
||||
|
||||
print("Testing different Risk-Reward ratios with optimized ICTStrategy...\n")
|
||||
|
||||
# Best params from optimization (you can tweak session/lookback etc. if you want)
|
||||
strategy = ICTStrategy(
|
||||
session="new_york", # Best was New York
|
||||
session="london",
|
||||
lookback=7,
|
||||
ob_max_age=20, # Best was 20
|
||||
ob_max_age=50,
|
||||
atr_mult=2.5,
|
||||
use_liquidity_sweep=False, # Best was False
|
||||
use_liquidity_sweep=True,
|
||||
sweep_lookback=5,
|
||||
)
|
||||
|
||||
total_start = time.perf_counter()
|
||||
|
||||
for rr in [1.0, 1.5, 2.0, 2.5, 3.0]:
|
||||
t0 = time.perf_counter()
|
||||
|
||||
trades = run_backtest(candles_5m, strategy, 10000, risk_reward=rr)
|
||||
|
||||
elapsed = time.perf_counter() - t0
|
||||
|
||||
if not trades:
|
||||
print(f"RR={rr}: No trades")
|
||||
print(f"RR={rr}: No trades ({elapsed:.2f}s)")
|
||||
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
|
||||
wr = len(winners) / len(trades) * 100
|
||||
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")
|
||||
print(
|
||||
f"RR={rr}: Trades={len(trades)}, WR={wr:.1f}%, PnL={total_pnl:.2f}, "
|
||||
f"AvgW={avg_win:.3f}, AvgL={avg_loss:.3f}, Time={elapsed:.2f}s"
|
||||
)
|
||||
|
||||
total_elapsed = time.perf_counter() - total_start
|
||||
print(f"Total run time: {total_elapsed:.2f}s")
|
||||
|
||||
Reference in New Issue
Block a user