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

Bug Cleanup

Browse Source
This commit is contained in:
moen0
2026-05-11 19:14:32 +02:00
parent 130da68f4e
commit 473fe3bd88
12 changed files with 348 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/routes.py
+81 -41
View File
@@ -10,7 +10,7 @@ from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from indicators.sessions import set_timezone
BACKEND_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
if BACKEND_DIR not in sys.path:
sys.path.insert(0, BACKEND_DIR)
@@ -141,13 +141,29 @@ def _get_candles_for_timeframe(dataset_id, timeframe):
def _build_strategy(
session, lookback, ob_age, atr_mult, use_fvg, use_ob,
proximity_pct, sweep, sweep_lookback,
min_gap_size, impulse_multiplier, require_unmitigated_fvg,
require_bos_confluence, min_ob_size, require_fvg_ob_confluence,
asian_sweep_only, day_filter,
use_break_even=False, be_trigger_rr=1.0,
use_partial_tp=False, partial_tp_rr=1.0, partial_tp_percent=50.0,
session="new_york",
lookback=5,
ob_age=50,
atr_mult=1.5,
use_fvg=True,
use_ob=True,
proximity_pct=0.3,
sweep=True,
sweep_lookback=10,
min_gap_size=0.0,
impulse_multiplier=0.0,
require_unmitigated_fvg=True,
require_bos_confluence=False,
min_ob_size=0.0,
require_fvg_ob_confluence=False,
asian_sweep_only=False,
day_filter=None,
use_break_even=False,
be_trigger_rr=1.0,
use_partial_tp=False,
partial_tp_rr=1.0,
partial_tp_percent=50.0,
timezone="est",
):
return ICTStrategy(
session=session,
@@ -172,6 +188,7 @@ def _build_strategy(
use_partial_tp=use_partial_tp,
partial_tp_rr=partial_tp_rr,
partial_tp_percent=partial_tp_percent,
timezone=timezone,
)
@@ -189,12 +206,28 @@ def _trade_payload(trade):
}
def _stats_payload(trades, rr):
def _stats_payload(trades, rr, starting_balance=10000.0):
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]
partial_tp_trades = [t for t in trades if getattr(t, "partial_tp_taken", False)]
partial_tp_realized_total = sum(float(getattr(t, "partial_tp_realized_pnl", 0.0) or 0.0) for t in partial_tp_trades)
pnls = [t.pnl for t in trades]
returns = [(p / starting_balance) for p in pnls] if starting_balance > 0 else []
mean_return = (sum(returns) / len(returns)) if returns else 0.0
variance = (sum((r - mean_return) ** 2 for r in returns) / len(returns)) if returns else 0.0
std_dev = variance ** 0.5
sharpe_ratio = ((mean_return / std_dev) * (len(returns) ** 0.5)) if std_dev > 0 else 0.0
equity_points = _build_equity_points(trades, starting_balance=starting_balance)
peak = equity_points[0] if equity_points else starting_balance
max_drawdown_pct = 0.0
for value in equity_points:
if value > peak:
peak = value
drawdown_pct = ((peak - value) / peak) * 100 if peak > 0 else 0.0
if drawdown_pct > max_drawdown_pct:
max_drawdown_pct = drawdown_pct
return {
"total_trades": len(trades),
"winners": len(winners),
@@ -208,6 +241,8 @@ def _stats_payload(trades, rr):
"partial_tp_rate": (len(partial_tp_trades) / len(trades) * 100) if trades else 0,
"partial_tp_realized_total": partial_tp_realized_total,
"partial_tp_realized_avg": (partial_tp_realized_total / len(partial_tp_trades)) if partial_tp_trades else 0,
"sharpe_ratio": round(sharpe_ratio, 6),
"max_drawdown_pct": round(max_drawdown_pct, 6),
}
@@ -310,7 +345,7 @@ def _risk_metrics(trades, starting_balance=10000.0):
sortino = (mean_pnl / downside_dev) * (trade_count ** 0.5) if downside_dev > 0 else 0.0
equity_points = _build_equity_points(trades, starting_balance=starting_balance)
peak = equity_points[0]
peak = equity_points[0] if equity_points else starting_balance
max_drawdown_pct = 0.0
for value in equity_points:
if value > peak:
@@ -319,8 +354,9 @@ def _risk_metrics(trades, starting_balance=10000.0):
if drawdown_pct > max_drawdown_pct:
max_drawdown_pct = drawdown_pct
calmar = (net_pnl / max_drawdown_pct) if max_drawdown_pct > 0 else 0.0
recovery = (net_pnl / max_drawdown_pct) if max_drawdown_pct > 0 else 0.0
calmar = ((net_pnl / starting_balance) * 100 / max_drawdown_pct) if max_drawdown_pct > 0 else 0.0
drawdown_amount = starting_balance * (max_drawdown_pct / 100) if max_drawdown_pct > 0 else 0.0
recovery = (net_pnl / drawdown_amount) if drawdown_amount > 0 else 0.0
if trade_count < 80:
trade_score = max(0.0, trade_count / 80)
@@ -524,24 +560,22 @@ def get_backtest(
max_consecutive_losses: int = 0,
):
dataset_id = _resolve_dataset(dataset)
if "MT5" in dataset.upper():
set_timezone("mt5")
else:
set_timezone("est")
timezone = "mt5" if "MT5" in dataset.upper() else "est"
candles = _get_candles_for_timeframe(dataset_id, timeframe)
strategy = _build_strategy(
session=session, lookback=lookback, ob_age=ob_age, atr_mult=atr_mult,
use_fvg=use_fvg, use_ob=use_ob, proximity_pct=proximity_pct,
sweep=sweep, sweep_lookback=sweep_lookback,
min_gap_size=min_gap_size, impulse_multiplier=impulse_multiplier,
require_unmitigated_fvg=require_unmitigated_fvg,
require_bos_confluence=require_bos_confluence,
min_ob_size=min_ob_size, require_fvg_ob_confluence=require_fvg_ob_confluence,
asian_sweep_only=asian_sweep_only, day_filter=day_filter,
use_break_even=use_break_even, be_trigger_rr=be_trigger_rr,
use_partial_tp=use_partial_tp, partial_tp_rr=partial_tp_rr, partial_tp_percent=partial_tp_percent,
)
session=session, lookback=lookback, ob_age=ob_age, atr_mult=atr_mult,
use_fvg=use_fvg, use_ob=use_ob, proximity_pct=proximity_pct,
sweep=sweep, sweep_lookback=sweep_lookback,
min_gap_size=min_gap_size, impulse_multiplier=impulse_multiplier,
require_unmitigated_fvg=require_unmitigated_fvg,
require_bos_confluence=require_bos_confluence,
min_ob_size=min_ob_size, require_fvg_ob_confluence=require_fvg_ob_confluence,
asian_sweep_only=asian_sweep_only, day_filter=day_filter,
use_break_even=use_break_even, be_trigger_rr=be_trigger_rr,
use_partial_tp=use_partial_tp, partial_tp_rr=partial_tp_rr, partial_tp_percent=partial_tp_percent,
timezone=timezone,
)
trades = run_backtest(
candles, strategy, 10000, risk_reward=rr,
max_daily_loss=max_daily_loss,
@@ -551,7 +585,7 @@ def get_backtest(
return {
"trades": [_trade_payload(t) for t in trades],
"candle_times": [c.time_open.isoformat() for c in candles],
"stats": _stats_payload(trades, rr),
"stats": _stats_payload(trades, rr, starting_balance=10000.0),
}
@@ -585,6 +619,7 @@ def backtest_monte_carlo(req: MonteCarloRequest):
@app.post("/api/optimize")
def get_optimize(req: OptimizeRequest):
dataset_id = _resolve_dataset(req.dataset)
timezone = "mt5" if "MT5" in dataset_id.upper() else "est"
candles = _get_candles_for_timeframe(dataset_id, req.timeframe)
session_list = req.sessions
@@ -709,6 +744,7 @@ def get_optimize(req: OptimizeRequest):
use_partial_tp=params["use_partial_tp"],
partial_tp_rr=params["partial_tp_rr"],
partial_tp_percent=params["partial_tp_percent"],
timezone=timezone,
)
trades = run_backtest(candles, strategy, 10000, risk_reward=params["rr"])
@@ -815,6 +851,7 @@ def get_optimize_monte_carlo(
ruin_drawdown_pct: float = Query(default=20.0, ge=0.0, le=100.0),
):
dataset_id = _resolve_dataset(dataset)
timezone = "mt5" if "MT5" in dataset_id.upper() else "est"
candles = _get_candles_for_timeframe(dataset_id, timeframe)
strategy = _build_strategy(
@@ -840,6 +877,7 @@ def get_optimize_monte_carlo(
partial_tp_rr=partial_tp_rr,
partial_tp_percent=partial_tp_percent,
day_filter=None,
timezone=timezone,
)
trades = run_backtest(candles, strategy, 10000, risk_reward=rr)
trade_r_multiples = [getattr(t, "r_multiple", 0.0) for t in trades]
@@ -948,20 +986,22 @@ def stream_backtest(
max_consecutive_losses: int = 0,
):
dataset_id = _resolve_dataset(dataset)
timezone = "mt5" if "MT5" in dataset_id.upper() else "est"
candles = _get_candles_for_timeframe(dataset_id, timeframe)
strategy = _build_strategy(
session=session, lookback=lookback, ob_age=ob_age, atr_mult=atr_mult,
use_fvg=use_fvg, use_ob=use_ob, proximity_pct=proximity_pct,
sweep=sweep, sweep_lookback=sweep_lookback,
min_gap_size=min_gap_size, impulse_multiplier=impulse_multiplier,
require_unmitigated_fvg=require_unmitigated_fvg,
require_bos_confluence=require_bos_confluence,
min_ob_size=min_ob_size, require_fvg_ob_confluence=require_fvg_ob_confluence,
asian_sweep_only=asian_sweep_only, day_filter=day_filter,
use_break_even=use_break_even, be_trigger_rr=be_trigger_rr,
use_partial_tp=use_partial_tp, partial_tp_rr=partial_tp_rr, partial_tp_percent=partial_tp_percent,
)
session=session, lookback=lookback, ob_age=ob_age, atr_mult=atr_mult,
use_fvg=use_fvg, use_ob=use_ob, proximity_pct=proximity_pct,
sweep=sweep, sweep_lookback=sweep_lookback,
min_gap_size=min_gap_size, impulse_multiplier=impulse_multiplier,
require_unmitigated_fvg=require_unmitigated_fvg,
require_bos_confluence=require_bos_confluence,
min_ob_size=min_ob_size, require_fvg_ob_confluence=require_fvg_ob_confluence,
asian_sweep_only=asian_sweep_only, day_filter=day_filter,
use_break_even=use_break_even, be_trigger_rr=be_trigger_rr,
use_partial_tp=use_partial_tp, partial_tp_rr=partial_tp_rr, partial_tp_percent=partial_tp_percent,
timezone=timezone,
)
def _sse(data):
return f"data: {json.dumps(data)}\n\n"
@@ -993,7 +1033,7 @@ def stream_backtest(
yield _sse({
"type": "trade",
"trade": _trade_payload(trade),
"stats": _stats_payload(streamed_trades, rr),
"stats": _stats_payload(streamed_trades, rr, starting_balance=10000.0),
"processed_candles": event["processed_candles"],
"total_candles": event["total_candles"],
})
@@ -1002,7 +1042,7 @@ def stream_backtest(
yield _sse({
"type": "done",
"trades": [_trade_payload(t) for t in streamed_trades],
"stats": _stats_payload(streamed_trades, rr),
"stats": _stats_payload(streamed_trades, rr, starting_balance=10000.0),
"duration_ms": round(duration_ms, 1),
"candle_times": [c.time_open.isoformat() for c in candles],
})
backend/data/loader.py
-1
View File
@@ -1,4 +1,3 @@
print("File is running")
import pandas as pd
from data.model import Candle
backend/data/model.py
+2
View File
@@ -28,3 +28,5 @@ class Trade:
exit_price: float
pnl: float
r_multiple: float = 0.0
partial_tp_taken: bool = False
partial_tp_realized_pnl: float = 0.0
backend/engine/backtester.py
+121 -4
View File
@@ -30,18 +30,66 @@ def _apply_break_even_if_triggered(position, candle, strategy):
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_tp_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
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)
if lot_size <= 0:
return
partial_pct = min(partial_pct, 100.0)
partial_lot = lot_size * (partial_pct / 100.0)
if partial_lot <= 0:
return
price_move = (trigger_price - entry) if is_long else (entry - trigger_price)
partial_pnl = price_move * partial_lot
position["lot_size"] = max(lot_size - partial_lot, 0.0)
position["partial_tp_taken"] = True
position["partial_tp_realized_pnl"] = position.get("partial_tp_realized_pnl", 0.0) + partial_pnl
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
consecutive_losses = 0
daily_pnl = defaultdict(float)
equity = float(starting_balance)
if hasattr(strategy, "prepare"):
strategy.prepare(candles)
for i, candle in enumerate(candles):
if position:
_apply_break_even_if_triggered(position, candle, strategy)
_apply_partial_tp_if_triggered(position, candle, strategy)
is_long = position["direction"] == "long"
sl, tp = position["stop_loss"], position["take_profit"]
@@ -53,7 +101,8 @@ def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
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)
pnl = price_move * lot_size
partial_pnl = float(position.get("partial_tp_realized_pnl", 0.0) or 0.0)
pnl = (price_move * lot_size) + partial_pnl
risk_distance = max(position.get("risk_distance", 0.0), 1e-12)
r_multiple = price_move / risk_distance
@@ -65,8 +114,11 @@ def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
exit_price=exit_price,
pnl=pnl,
r_multiple=r_multiple,
partial_tp_taken=bool(position.get("partial_tp_taken", False)),
partial_tp_realized_pnl=partial_pnl,
))
position = None
equity += pnl
if pnl <= 0:
consecutive_losses += 1
@@ -76,6 +128,8 @@ def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
daily_pnl[candle.time_open.date()] += pnl
if position is None:
if equity <= 0:
continue
if max_consecutive_losses > 0 and consecutive_losses >= max_consecutive_losses:
continue
if max_daily_loss > 0:
@@ -99,7 +153,7 @@ def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
):
continue
risk_amount = starting_balance * (risk_pct / 100)
risk_amount = equity * (risk_pct / 100)
if risk_amount <= 0 or not math.isfinite(risk_amount):
continue
@@ -118,8 +172,35 @@ def run_backtest(candles, strategy, starting_balance, risk_reward=1.0,
"risk_distance": sl_distance,
"lot_size": lot_size,
"break_even_armed": False,
"partial_tp_taken": False,
"partial_tp_realized_pnl": 0.0,
}
if position and candles:
last_candle = candles[-1]
is_long = position["direction"] == "long"
exit_price = last_candle.close
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)
partial_pnl = float(position.get("partial_tp_realized_pnl", 0.0) or 0.0)
pnl = (price_move * lot_size) + partial_pnl
risk_distance = max(position.get("risk_distance", 0.0), 1e-12)
r_multiple = price_move / risk_distance
trades.append(Trade(
enter_time=position["enter_time"],
enter_price=position["entry_price"],
direction=position["direction"],
exit_time=last_candle.time_open,
exit_price=exit_price,
pnl=pnl,
r_multiple=r_multiple,
partial_tp_taken=bool(position.get("partial_tp_taken", False)),
partial_tp_realized_pnl=partial_pnl,
))
equity += pnl
daily_pnl[last_candle.time_open.date()] += pnl
return trades
@@ -129,6 +210,7 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
consecutive_losses = 0
daily_pnl = defaultdict(float)
total = len(candles)
equity = float(starting_balance)
if hasattr(strategy, "prepare"):
strategy.prepare(candles)
@@ -143,6 +225,7 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
if position:
_apply_break_even_if_triggered(position, candle, strategy)
_apply_partial_tp_if_triggered(position, candle, strategy)
is_long = position["direction"] == "long"
sl, tp = position["stop_loss"], position["take_profit"]
@@ -154,7 +237,8 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
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)
pnl = price_move * lot_size
partial_pnl = float(position.get("partial_tp_realized_pnl", 0.0) or 0.0)
pnl = (price_move * lot_size) + partial_pnl
risk_distance = max(position.get("risk_distance", 0.0), 1e-12)
r_multiple = price_move / risk_distance
@@ -166,8 +250,11 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
exit_price=exit_price,
pnl=pnl,
r_multiple=r_multiple,
partial_tp_taken=bool(position.get("partial_tp_taken", False)),
partial_tp_realized_pnl=partial_pnl,
)
position = None
equity += pnl
if pnl <= 0:
consecutive_losses += 1
@@ -179,6 +266,8 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
yield {"type": "trade", "trade": trade, "processed_candles": i, "total_candles": total}
if position is None:
if equity <= 0:
continue
if max_consecutive_losses > 0 and consecutive_losses >= max_consecutive_losses:
continue
if max_daily_loss > 0:
@@ -202,7 +291,7 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
):
continue
risk_amount = starting_balance * (risk_pct / 100)
risk_amount = equity * (risk_pct / 100)
if risk_amount <= 0 or not math.isfinite(risk_amount):
continue
@@ -221,6 +310,34 @@ def run_backtest_stream(candles, strategy, starting_balance, risk_reward=1.0,
"risk_distance": sl_distance,
"lot_size": lot_size,
"break_even_armed": False,
"partial_tp_taken": False,
"partial_tp_realized_pnl": 0.0,
}
if position and candles:
last_candle = candles[-1]
is_long = position["direction"] == "long"
exit_price = last_candle.close
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)
partial_pnl = float(position.get("partial_tp_realized_pnl", 0.0) or 0.0)
pnl = (price_move * lot_size) + partial_pnl
risk_distance = max(position.get("risk_distance", 0.0), 1e-12)
r_multiple = price_move / risk_distance
trade = Trade(
enter_time=position["enter_time"],
enter_price=position["entry_price"],
direction=position["direction"],
exit_time=last_candle.time_open,
exit_price=exit_price,
pnl=pnl,
r_multiple=r_multiple,
partial_tp_taken=bool(position.get("partial_tp_taken", False)),
partial_tp_realized_pnl=partial_pnl,
)
equity += pnl
daily_pnl[last_candle.time_open.date()] += pnl
yield {"type": "trade", "trade": trade, "processed_candles": total, "total_candles": total}
yield {"type": "done", "total_candles": total}
backend/indicators/sessions.py
+21 -14
View File
@@ -22,36 +22,42 @@ SESSIONS_MT5 = {
_active_sessions = SESSIONS_EST
def get_sessions_for_tz(tz="est"):
if tz and tz.lower() in ("mt5", "utc+2", "server"):
return SESSIONS_MT5
return SESSIONS_EST
def set_timezone(tz="est"):
global _active_sessions
if tz.lower() in ("mt5", "utc+2", "server"):
_active_sessions = SESSIONS_MT5
else:
_active_sessions = SESSIONS_EST
_active_sessions = get_sessions_for_tz(tz)
def in_session(candle_time, session_name):
def in_session(candle_time, session_name, sessions_map=None):
if session_name == "all":
return True
if session_name not in _active_sessions:
active = sessions_map or _active_sessions
if session_name not in active:
return True
t = candle_time.time()
start, end = _active_sessions[session_name]
start, end = active[session_name]
if start > end:
return t >= start or t < end
return start <= t < end
def get_session(candle_time):
for name in _active_sessions:
if in_session(candle_time, name):
def get_session(candle_time, sessions_map=None):
active = sessions_map or _active_sessions
for name in active:
if in_session(candle_time, name, sessions_map=active):
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 filter_by_session(candles, session_name, sessions_map=None):
active = sessions_map or _active_sessions
return [c for c in candles if in_session(c.time_open, session_name, sessions_map=active)]
def in_day_filter(candle_time, allowed_days):
@@ -60,8 +66,9 @@ def in_day_filter(candle_time, allowed_days):
return candle_time.weekday() in allowed_days
def get_asian_range(candles):
asian = filter_by_session(candles, "asian")
def get_asian_range(candles, sessions_map=None):
active = sessions_map or _active_sessions
asian = filter_by_session(candles, "asian", sessions_map=active)
if not asian:
return None
return {
backend/optimize_categorical.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/gbpjpy_jan.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/strategies/ict_strategy.py
+5 -3
View File
@@ -3,7 +3,7 @@ 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 indicators.sessions import in_session, in_day_filter, get_asian_range
from indicators.sessions import in_session, in_day_filter, get_asian_range, get_sessions_for_tz
from collections import defaultdict
@@ -33,6 +33,7 @@ class ICTStrategy:
use_partial_tp=False,
partial_tp_rr=1.0,
partial_tp_percent=50.0,
timezone="est",
):
self.lookback = lookback
self.atr_mult = atr_mult
@@ -58,6 +59,7 @@ class ICTStrategy:
self.use_partial_tp = use_partial_tp
self.partial_tp_rr = partial_tp_rr
self.partial_tp_percent = partial_tp_percent
self.sessions_map = get_sessions_for_tz(timezone)
self.swings = []
self.structure = []
@@ -88,7 +90,7 @@ class ICTStrategy:
for c in candles:
daily[c.time_open.date()].append(c)
for date, day_candles in daily.items():
ar = get_asian_range(day_candles)
ar = get_asian_range(day_candles, sessions_map=self.sessions_map)
if ar:
self.asian_ranges[date] = ar
@@ -204,7 +206,7 @@ class ICTStrategy:
candle = candles[index]
if not in_session(candle.time_open, self.session):
if not in_session(candle.time_open, self.session, sessions_map=self.sessions_map):
self.recent_sweep = None
return None