gavindiaz/fx-quant
1
0
Fork 0
mirror of https://github.com/BrentNeale1/fx-quant.git synced 2026-08-07 15:47:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

Phase 1 complete: S7-S9 Smart Money strategies, expanded pair testing, consolidated scorecard

Browse Source 
- S7 Liquidity Sweep: built, tested across 6 pairs, tight SL (1.0 ATR) on GBP_JPY
  is Phase 2 candidate (107 trades, OOS PF 1.39, gen ratio 1.81)
- S8 Order Block: built, tested on GBP_JPY (watchlist, 32 trades, OOS PF 1.55)
- S9 London Session: built, tested across 8 pairs with filter experiments
  GBP_USD (OOS PF 1.45) and GBP_AUD filtered (OOS PF 1.94) advance to Phase 2
- Added OBV indicator to technical.py
- Added GBP_NZD to engine spread/pip config
- Standalone OANDA fetcher (bypasses Supabase dependency)
- Fetched EUR_GBP, EUR_USD, GBP_NZD H1 data (2021-2023)
- Consolidated STRATEGY_LEARNINGS.md with full Phase 1 scorecard and 11 design principles
- Phase 2 roster: S7/GBP_JPY, S9/GBP_USD, S9F/GBP_AUD, S4-F/EUR_AUD, S3/GBP_JPY

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Brent Neale
2026-02-19 13:56:03 +10:00
parent 70a216d695
commit 39a6536284
83 changed files with 7737 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/analyze_s9_pair.py
+147
View File
@@ -0,0 +1,147 @@
"""Quick S9 trade analysis for a single pair — runs backtest and analyzes trade log."""
import os, sys, io
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
import numpy as np
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
from src.strategies_pkg.s9_london_session import S9_London_Session
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
pair = sys.argv[1] if len(sys.argv) > 1 else "GBP_AUD"
# Load and run
fp = os.path.join(PROCESSED_DIR, f"{pair}_H1.csv")
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
df = compute_all_indicators(df)
bt = Backtester(data=df, strategy=S9_London_Session(), pair=pair, starting_equity=100_000.0, htf_data=df.copy())
bt.run()
trades = bt.get_trade_log_df()
print(f"\n{'='*70}")
print(f"S9 TRADE ANALYSIS — {pair} ({len(trades)} trades)")
print(f"{'='*70}")
if len(trades) == 0:
print("No trades!")
sys.exit(0)
trades["win"] = trades["pnl_pips"] > 0
# 1. Exit reason breakdown
print("\n1. EXIT REASON BREAKDOWN")
for reason, grp in trades.groupby("exit_reason"):
n = len(grp)
wr = grp["win"].mean() * 100
avg_pnl = grp["pnl_pips"].mean()
print(f" {reason:<12} {n:>4} trades ({n/len(trades)*100:5.1f}%) | WR {wr:5.1f}% | Avg PnL {avg_pnl:+7.1f}p")
# 2. Direction
print("\n2. WIN RATE BY DIRECTION")
for d, grp in trades.groupby("signal_direction"):
print(f" {d:<6} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 3. Day of week
print("\n3. WIN RATE BY DAY OF WEEK")
trades["dow"] = pd.to_datetime(trades["timestamp"]).dt.day_name()
for day in ["Monday","Tuesday","Wednesday","Thursday","Friday"]:
grp = trades[trades["dow"] == day]
if len(grp) > 0:
print(f" {day:<10} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 4. Hold time
print("\n4. HOLD TIME (minutes)")
winners = trades[trades["win"]]
losers = trades[~trades["win"]]
print(f" Winners: avg {winners['hold_time_minutes'].mean():.0f}m, median {winners['hold_time_minutes'].median():.0f}m")
print(f" Losers: avg {losers['hold_time_minutes'].mean():.0f}m, median {losers['hold_time_minutes'].median():.0f}m")
# 5. Confluence
print("\n5. CONFLUENCE SCORE")
for score, grp in trades.groupby("confluence_score"):
print(f" Score {score}: {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 6. Entry hour
print("\n6. ENTRY HOUR")
for h, grp in trades.groupby("hour_of_day"):
print(f" Hour {h:>2}: {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 7. RSI zones
print("\n7. RSI AT ENTRY")
bins = [0, 30, 40, 50, 60, 70, 100]
labels = ["<30", "30-40", "40-50", "50-60", "60-70", "70+"]
trades["rsi_bin"] = pd.cut(trades["rsi_at_entry"], bins=bins, labels=labels, include_lowest=True)
for b in labels:
grp = trades[trades["rsi_bin"] == b]
if len(grp) > 0:
print(f" RSI {b:<6} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# RSI neutral vs momentum
neutral = trades[(trades["rsi_at_entry"] >= 40) & (trades["rsi_at_entry"] <= 60)]
momentum = trades[(trades["rsi_at_entry"] < 40) | (trades["rsi_at_entry"] > 60)]
print(f" RSI 40-60 (neutral): {len(neutral):>4} trades, WR {neutral['win'].mean()*100:5.1f}%, Avg PnL {neutral['pnl_pips'].mean():+7.1f}p")
print(f" RSI outside (momentum): {len(momentum):>4} trades, WR {momentum['win'].mean()*100:5.1f}%, Avg PnL {momentum['pnl_pips'].mean():+7.1f}p")
# 8. ADX buckets
print("\n8. ADX AT ENTRY")
adx_bins = [0, 20, 25, 30, 40, 100]
adx_labels = ["<20", "20-25", "25-30", "30-40", "40+"]
trades["adx_bin"] = pd.cut(trades["adx_at_entry"], bins=adx_bins, labels=adx_labels, include_lowest=True)
for b in adx_labels:
grp = trades[trades["adx_bin"] == b]
if len(grp) > 0:
print(f" ADX {b:<6} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 9. EMA50 distance
print("\n9. DISTANCE FROM EMA50 (absolute pips)")
trades["ema_dist_abs"] = trades["distance_from_ema50_pips"].abs()
dist_bins = [0, 20, 40, 60, 100, 500]
dist_labels = ["0-20", "20-40", "40-60", "60-100", "100+"]
trades["dist_bin"] = pd.cut(trades["ema_dist_abs"], bins=dist_bins, labels=dist_labels, include_lowest=True)
for b in dist_labels:
grp = trades[trades["dist_bin"] == b]
if len(grp) > 0:
print(f" {b:<8} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 10. Candle body ratio
print("\n10. CANDLE BODY RATIO")
br_bins = [0, 0.3, 0.5, 0.7, 1.01]
br_labels = ["<0.3", "0.3-0.5", "0.5-0.7", "0.7+"]
trades["br_bin"] = pd.cut(trades["candle_body_ratio"], bins=br_bins, labels=br_labels, include_lowest=True)
for b in br_labels:
grp = trades[trades["br_bin"] == b]
if len(grp) > 0:
print(f" {b:<8} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# 11. Spread as % of avg win
avg_win_pips = winners["pnl_pips"].mean() if len(winners) > 0 else 0
avg_spread = trades["spread_at_entry"].mean()
print(f"\n11. SPREAD IMPACT")
print(f" Avg spread: {avg_spread:.1f} pips")
print(f" Avg win: {avg_win_pips:.1f} pips")
print(f" Spread/win: {avg_spread/avg_win_pips*100:.1f}%" if avg_win_pips > 0 else " N/A")
# 12. Yearly breakdown
print("\n12. YEARLY BREAKDOWN")
trades["year"] = pd.to_datetime(trades["timestamp"]).dt.year
for y, grp in trades.groupby("year"):
gw = grp[grp["pnl_pips"] > 0]["pnl_pips"].sum()
gl = abs(grp[grp["pnl_pips"] < 0]["pnl_pips"].sum())
pf = gw / gl if gl > 0 else 0
print(f" {y}: {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, PF {pf:.2f}, PnL {grp['pnl_pips'].sum():+8.1f}p")
# 13. Session (if available)
if "session" in trades.columns:
print("\n13. SESSION")
for s, grp in trades.groupby("session"):
print(f" {s:<10} {len(grp):>4} trades, WR {grp['win'].mean()*100:5.1f}%, Avg PnL {grp['pnl_pips'].mean():+7.1f}p")
# Save trade log
out = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1", f"S9_{pair}_FULL_trades.csv")
trades.to_csv(out, index=False)
print(f"\nTrade log saved: {out}")
src/backtester/engine.py
+2
View File
@@ -26,6 +26,7 @@ SPREAD_PIPS = {
"EUR_GBP": 2.0, "GBP_JPY": 2.5,
"USD_JPY": 1.5, "GBP_CAD": 2.5,
"EUR_CAD": 2.5, "EUR_NZD": 2.5,
"GBP_NZD": 2.5,
}
# Pip value per pair
@@ -33,6 +34,7 @@ PIP_SIZE = {
"EUR_USD": 0.0001, "GBP_USD": 0.0001, "EUR_AUD": 0.0001,
"GBP_AUD": 0.0001, "EUR_GBP": 0.0001, "GBP_CAD": 0.0001,
"EUR_CAD": 0.0001, "EUR_NZD": 0.0001,
"GBP_NZD": 0.0001,
"USD_JPY": 0.01, "GBP_JPY": 0.01,
}
src/fetch_oanda_standalone.py
+147
View File
@@ -0,0 +1,147 @@
"""
Standalone OANDA data fetcher — no Supabase dependency.
Fetches H1 candles and saves to data/processed/ in the same CSV format
as existing data files (timestamp, open, high, low, close, volume).
Usage:
python src/fetch_oanda_standalone.py
"""
import os
import sys
import time
from datetime import datetime, timezone, timedelta
from pathlib import Path
import requests
import pandas as pd
# Load OANDA key from config/.env
ROOT = Path(__file__).resolve().parent.parent
env_path = ROOT / "config" / ".env"
if env_path.exists():
for line in env_path.read_text().splitlines():
line = line.strip()
if line and not line.startswith("#") and "=" in line:
k, v = line.split("=", 1)
os.environ.setdefault(k.strip(), v.strip())
API_KEY = os.getenv("OANDA_API_KEY")
ENV = os.getenv("OANDA_ENV", "practice")
BASE = (
"https://api-fxpractice.oanda.com"
if ENV == "practice"
else "https://api-fxtrade.oanda.com"
)
HEADERS = {"Authorization": f"Bearer {API_KEY}"}
def fetch_candles_chunk(instrument, granularity, from_dt, to_dt):
"""Fetch up to 5000 candles between from_dt and to_dt."""
params = {
"granularity": granularity,
"from": from_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
"to": to_dt.strftime("%Y-%m-%dT%H:%M:%SZ"),
}
url = f"{BASE}/v3/instruments/{instrument}/candles"
r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()
return r.json().get("candles", [])
def fetch_all_candles(instrument, granularity, start_dt, end_dt):
"""Paginate through OANDA history in chunks of ~5000 candles."""
gran_minutes = {
"M1": 1, "M5": 5, "M15": 15, "M30": 30,
"H1": 60, "H4": 240, "D": 1440, "W": 10080,
}
minutes = gran_minutes.get(granularity, 60)
chunk_duration = timedelta(minutes=minutes * 4999)
all_candles = []
cursor = start_dt
chunk_num = 0
while cursor < end_dt:
chunk_end = min(cursor + chunk_duration, end_dt)
chunk_num += 1
print(f" Chunk {chunk_num}: {cursor.strftime('%Y-%m-%d %H:%M')} -> "
f"{chunk_end.strftime('%Y-%m-%d %H:%M')} ...", end=" ", flush=True)
candles = fetch_candles_chunk(instrument, granularity, cursor, chunk_end)
print(f"{len(candles)} candles", flush=True)
if candles:
all_candles.extend(candles)
last_time = pd.to_datetime(candles[-1]["time"])
cursor = last_time.to_pydatetime().replace(tzinfo=timezone.utc) + timedelta(minutes=minutes)
else:
cursor = chunk_end
time.sleep(0.5)
print(f" Total raw candles fetched: {len(all_candles)}", flush=True)
if not all_candles:
return pd.DataFrame()
# Convert to DataFrame
rows = []
for c in all_candles:
mid = c["mid"]
rows.append({
"timestamp": c["time"],
"open": float(mid["o"]),
"high": float(mid["h"]),
"low": float(mid["l"]),
"close": float(mid["c"]),
"volume": c.get("volume", 0),
})
df = pd.DataFrame(rows)
df["timestamp"] = pd.to_datetime(df["timestamp"])
df = df.drop_duplicates(subset=["timestamp"], keep="last")
df = df.sort_values("timestamp").reset_index(drop=True)
print(f" After dedup: {len(df)} candles", flush=True)
print(f" Range: {df['timestamp'].iloc[0]} -> {df['timestamp'].iloc[-1]}", flush=True)
return df
def main():
if not API_KEY:
print("ERROR: No OANDA_API_KEY found in config/.env")
sys.exit(1)
# Match existing data range: 2020-12-31 to 2023-08-31
start_dt = datetime(2020, 12, 31, 0, 0, tzinfo=timezone.utc)
end_dt = datetime(2023, 8, 31, 23, 59, tzinfo=timezone.utc)
pairs = ["EUR_GBP", "EUR_USD", "GBP_NZD"]
granularity = "H1"
out_dir = ROOT / "data" / "processed"
out_dir.mkdir(parents=True, exist_ok=True)
for pair in pairs:
out_file = out_dir / f"{pair}_{granularity}.csv"
if out_file.exists():
print(f"\n{pair} already exists at {out_file}, skipping.")
continue
print(f"\n{'='*60}")
print(f"Fetching {pair} / {granularity} ({start_dt.date()} -> {end_dt.date()})")
print(f"{'='*60}")
df = fetch_all_candles(pair, granularity, start_dt, end_dt)
if df.empty:
print(f" No data for {pair}. Skipping.")
continue
df.to_csv(out_file, index=False)
print(f" Saved: {out_file} ({len(df)} rows)")
print("\nDone.")
if __name__ == "__main__":
main()
src/indicators/technical.py
+14
View File
@@ -142,6 +142,17 @@ def session_vwap_bands(df: pd.DataFrame, session_start_hour: int = 8):
}, index=df.index)
# ---------------------------------------------------------------------------
# On-Balance Volume (OBV)
# ---------------------------------------------------------------------------
def obv(df: pd.DataFrame) -> pd.Series:
"""Cumulative OBV: adds volume on up-closes, subtracts on down-closes."""
direction = np.where(df["close"] > df["close"].shift(1), 1,
np.where(df["close"] < df["close"].shift(1), -1, 0))
return (df["volume"] * direction).cumsum()
# ---------------------------------------------------------------------------
# Swing High / Low Detection
# ---------------------------------------------------------------------------
@@ -403,4 +414,7 @@ def compute_all_indicators(df: pd.DataFrame) -> pd.DataFrame:
df["is_swing_high"] = swing_highs(df)
df["is_swing_low"] = swing_lows(df)
# On-Balance Volume
df["obv"] = obv(df)
return df
src/run_s789_comparison.py
+267
View File
@@ -0,0 +1,267 @@
"""
Run S7, S8, S9 (Smart Money strategies) comparison.
Uses last 6 months of available data with 70/30 train/test split.
Data range: ~March 2023 - August 2023 (last 6 months of processed data).
"""
import os
import sys
import json
import time
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
PAIRS = ["GBP_USD", "EUR_AUD", "GBP_JPY"]
# 6 months of data, 70/30 split
DATA_MONTHS = 6
TRAIN_RATIO = 0.70
def load_data(pair, tf, last_n_months=None):
"""Load and compute indicators. Optionally slice to last N months."""
fp = os.path.join(PROCESSED_DIR, f"{pair}_{tf}.csv")
if not os.path.exists(fp):
print(f" WARNING: {fp} not found, skipping")
return None
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
if last_n_months is not None and last_n_months > 0:
end_date = df.index[-1]
start_date = end_date - pd.DateOffset(months=last_n_months)
# Keep extra warmup bars (250) before the start for indicator computation
warmup_start = start_date - pd.DateOffset(days=60)
df = df[df.index >= warmup_start]
df = compute_all_indicators(df)
if last_n_months is not None and last_n_months > 0:
# Now trim to the actual date range (after indicators computed with warmup)
df = df[df.index >= start_date]
return df
def split_data(df, train_ratio=0.70):
"""Split data into train and test sets."""
split_idx = int(len(df) * train_ratio)
return df.iloc[:split_idx], df.iloc[split_idx:]
def run_backtest(strat_cls, data, htf_data, pair, label=""):
"""Run a single backtest and return report + trade log."""
strategy = strat_cls()
bt = Backtester(
data=data,
strategy=strategy,
pair=pair,
starting_equity=100_000.0,
htf_data=htf_data,
)
report = bt.run()
trade_log = bt.get_trade_log_df()
return report, trade_log
def run_variant(label, strat_cls, phase="FULL"):
"""Run a strategy variant across all pairs."""
print(f"\n{'=' * 60}")
print(f" {label} ({phase})")
print(f"{'=' * 60}")
all_reports = []
all_trades = []
for pair in PAIRS:
# Load H1 as primary timeframe for S7-S9
data = load_data(pair, "H1", last_n_months=DATA_MONTHS)
if data is None:
continue
# HTF data: use H1 itself (strategies use htf_row for trend context)
# The backtester passes the most recent fully-closed H1 bar as htf_row
htf_data = data.copy()
if phase == "TRAIN":
data, _ = split_data(data, TRAIN_RATIO)
htf_data, _ = split_data(htf_data, TRAIN_RATIO)
elif phase == "TEST":
_, data = split_data(data, TRAIN_RATIO)
_, htf_data = split_data(htf_data, TRAIN_RATIO)
print(f" {pair}: H1={len(data)} bars ({data.index[0].date()} to {data.index[-1].date()})", end=" ")
t0 = time.time()
report, trade_log = run_backtest(strat_cls, data, htf_data, pair, label)
elapsed = time.time() - t0
n = report.get("total_trades", 0)
print(f"-> {n} trades ({elapsed:.0f}s)")
all_reports.append(report)
if len(trade_log) > 0:
all_trades.append(trade_log)
# Aggregate results
total_trades = sum(r.get("total_trades", 0) for r in all_reports)
if total_trades == 0:
print(f" NO TRADES across any pair!")
return {"total_trades": 0, "reports": all_reports}
total_wins = sum(
r.get("total_trades", 0) * r.get("win_rate_pct", 0) / 100
for r in all_reports
)
total_pnl_pips = sum(r.get("total_pnl_pips", 0) for r in all_reports)
total_pnl_dollars = sum(r.get("total_pnl_dollars", 0) for r in all_reports)
# Combined trade log for aggregate PF
combined = pd.concat(all_trades, ignore_index=True) if all_trades else pd.DataFrame()
if len(combined) > 0:
gross_profit = combined.loc[combined["pnl_pips"] > 0, "pnl_pips"].sum()
gross_loss = abs(combined.loc[combined["pnl_pips"] < 0, "pnl_pips"].sum())
pf = gross_profit / gross_loss if gross_loss > 0 else 0
avg_win = combined.loc[combined["pnl_pips"] > 0, "pnl_pips"].mean() if (combined["pnl_pips"] > 0).any() else 0
avg_loss = abs(combined.loc[combined["pnl_pips"] < 0, "pnl_pips"].mean()) if (combined["pnl_pips"] < 0).any() else 0
avg_rr = avg_win / avg_loss if avg_loss > 0 else 0
max_dd = min(r.get("max_drawdown_pct", 0) for r in all_reports)
else:
pf = avg_win = avg_loss = avg_rr = max_dd = 0
win_rate = total_wins / total_trades * 100 if total_trades > 0 else 0
expectancy = total_pnl_pips / total_trades if total_trades > 0 else 0
print(f"\n --- {label} {phase} AGGREGATE ---")
print(f" Trades: {total_trades}")
print(f" Win Rate: {win_rate:.1f}%")
print(f" Avg RR: {avg_rr:.2f}")
print(f" Expectancy: {expectancy:.2f} pips")
print(f" Profit Factor: {pf:.2f}")
print(f" Worst Max DD: {max_dd:.2f}%")
print(f" Total PnL: {total_pnl_pips:.1f} pips / ${total_pnl_dollars:,.2f}")
print(f" Avg Win: {avg_win:.1f}p | Avg Loss: {avg_loss:.1f}p")
# Per-pair breakdown
for i, pair in enumerate(PAIRS):
if i < len(all_reports):
r = all_reports[i]
n = r.get("total_trades", 0)
wr = r.get("win_rate_pct", 0)
pfp = r.get("profit_factor", 0)
pnl = r.get("total_pnl_pips", 0)
dd = r.get("max_drawdown_pct", 0)
print(f" {pair}: {n} trades, WR {wr:.1f}%, PF {pfp:.2f}, PnL {pnl:.1f}p, DD {dd:.2f}%")
# Save trade logs
if len(combined) > 0:
os.makedirs(RESULTS_DIR, exist_ok=True)
csv_path = os.path.join(RESULTS_DIR, f"{label}_{phase}_trades.csv")
combined.to_csv(csv_path, index=False)
print(f" Trade log saved: {csv_path}")
return {
"total_trades": total_trades, "win_rate": round(win_rate, 2),
"avg_rr": round(avg_rr, 2), "expectancy": round(expectancy, 2),
"profit_factor": round(pf, 2), "max_dd": round(max_dd, 2),
"total_pnl_pips": round(total_pnl_pips, 1),
"total_pnl_dollars": round(total_pnl_dollars, 2),
"avg_win": round(avg_win, 1), "avg_loss": round(avg_loss, 1),
"reports": all_reports,
}
if __name__ == "__main__":
from src.strategies_pkg.s7_liquidity_sweep import S7_Liquidity_Sweep
from src.strategies_pkg.s8_order_block import S8_Order_Block
from src.strategies_pkg.s9_london_session import S9_London_Session
strategies = {
"S7": ("S7 (Liquidity Sweep Reversal)", S7_Liquidity_Sweep),
"S8": ("S8 (Order Block Retest)", S8_Order_Block),
"S9": ("S9 (London Session Gap)", S9_London_Session),
}
# Parse CLI args
variant = sys.argv[1].upper() if len(sys.argv) > 1 else "ALL"
results = {}
for key, (label, strat_cls) in strategies.items():
if variant != "ALL" and variant != key:
continue
print(f"\n{'#' * 70}")
print(f"# {label}")
print(f"{'#' * 70}")
# Full 6-month run
full = run_variant(key, strat_cls, phase="FULL")
# 70/30 split: train
train = run_variant(key, strat_cls, phase="TRAIN")
# 70/30 split: test (OOS)
test = run_variant(key, strat_cls, phase="TEST")
results[key] = {"full": full, "train": train, "test": test}
# Summary comparison
if len(results) > 1:
print(f"\n{'=' * 80}")
print("STRATEGY COMPARISON — FULL 6 MONTHS")
print(f"{'=' * 80}")
print(f"{'Metric':<20}", end="")
for k in results:
print(f"{k:>18}", end="")
print()
print("-" * (20 + 18 * len(results)))
for metric in ["total_trades", "win_rate", "avg_rr", "expectancy",
"profit_factor", "max_dd", "total_pnl_pips",
"total_pnl_dollars", "avg_win", "avg_loss"]:
print(f"{metric:<20}", end="")
for k in results:
v = results[k]["full"].get(metric, 0)
if isinstance(v, float):
print(f"{v:>18.2f}", end="")
else:
print(f"{v:>18}", end="")
print()
print(f"\n{'=' * 80}")
print("TRAIN vs TEST (70/30 Split)")
print(f"{'=' * 80}")
for k in results:
train_r = results[k]["train"]
test_r = results[k]["test"]
print(f"\n {k}:")
print(f" TRAIN: {train_r.get('total_trades',0)} trades, "
f"WR {train_r.get('win_rate',0):.1f}%, "
f"PF {train_r.get('profit_factor',0):.2f}, "
f"PnL {train_r.get('total_pnl_pips',0):.1f}p")
print(f" TEST: {test_r.get('total_trades',0)} trades, "
f"WR {test_r.get('win_rate',0):.1f}%, "
f"PF {test_r.get('profit_factor',0):.2f}, "
f"PnL {test_r.get('total_pnl_pips',0):.1f}p")
if train_r.get('profit_factor', 0) > 0 and test_r.get('profit_factor', 0) > 0:
gen_ratio = test_r['profit_factor'] / train_r['profit_factor']
print(f" Generalization Ratio (PF): {gen_ratio:.2f}")
# Save summary
os.makedirs(RESULTS_DIR, exist_ok=True)
summary = {}
for k, v in results.items():
summary[k] = {
phase: {mk: mv for mk, mv in data.items() if mk != "reports"}
for phase, data in v.items()
}
with open(os.path.join(RESULTS_DIR, "s789_comparison.json"), "w") as f:
json.dump(summary, f, indent=2)
print(f"\nSummary saved: {os.path.join(RESULTS_DIR, 's789_comparison.json')}")
src/run_s7_expanded.py
+187
View File
@@ -0,0 +1,187 @@
"""
S7 Liquidity Sweep — Expanded testing.
1. GBP_JPY: Full dataset (2021-2023) with 70/30 split
2. Other 5 pairs: Last 6 months with 70/30 split
- GBP_USD, EUR_AUD, GBP_AUD, EUR_USD, GBP_NZD
Uses the tight SL variant (1.0 ATR) which proved best on GBP_JPY 6-month.
"""
import os, sys, io, json, time
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
from src.strategies_pkg.s7_liquidity_sweep import S7_Liquidity_Sweep
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
os.makedirs(RESULTS_DIR, exist_ok=True)
TRAIN_RATIO = 0.70
def load_data(pair, last_n_months=None):
"""Load H1 data, optionally slice to last N months (with warmup)."""
fp = os.path.join(PROCESSED_DIR, f"{pair}_H1.csv")
if not os.path.exists(fp):
print(f" WARNING: {fp} not found, skipping")
return None
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
if last_n_months is not None and last_n_months > 0:
end_date = df.index[-1]
start_date = end_date - pd.DateOffset(months=last_n_months)
warmup_start = start_date - pd.DateOffset(days=60)
df = df[df.index >= warmup_start]
df = compute_all_indicators(df)
if last_n_months is not None and last_n_months > 0:
df = df[df.index >= start_date]
return df
def split_data(df):
split_idx = int(len(df) * TRAIN_RATIO)
return df.iloc[:split_idx], df.iloc[split_idx:]
def run_test(pair, data, htf_data, phase="FULL"):
if phase == "TRAIN":
data, _ = split_data(data)
htf_data, _ = split_data(htf_data)
elif phase == "TEST":
_, data = split_data(data)
_, htf_data = split_data(htf_data)
strategy = S7_Liquidity_Sweep()
bt = Backtester(data=data, strategy=strategy, pair=pair,
starting_equity=100_000.0, htf_data=htf_data)
report = bt.run()
trade_log = bt.get_trade_log_df()
return report, trade_log
def extract_metrics(report):
n = report.get("total_trades", 0)
wr = report.get("win_rate_pct", 0)
pf = report.get("profit_factor", 0)
pnl = report.get("total_pnl_pips", 0)
pnl_d = report.get("total_pnl_dollars", 0)
dd = report.get("max_drawdown_pct", 0)
aw = report.get("avg_win_pips", 0)
al = report.get("avg_loss_pips", 0)
rr = aw / al if al > 0 else 0
exp = pnl / n if n > 0 else 0
return {
"trades": n, "wr": round(wr, 1), "pf": round(pf, 2),
"pnl_pips": round(pnl, 1), "pnl_usd": round(pnl_d, 2),
"dd": round(dd, 2), "avg_win": round(aw, 1),
"avg_loss": round(al, 1), "rr": round(rr, 2),
"expectancy": round(exp, 2),
}
if __name__ == "__main__":
all_results = {}
# ---- TEST 1: GBP_JPY Full Dataset ----
pair = "GBP_JPY"
print(f"\n{'#' * 70}")
print(f"# S7 Tight — {pair} FULL DATASET (2021-2023)")
print(f"{'#' * 70}")
data = load_data(pair, last_n_months=None)
htf = data.copy()
print(f" Data: {len(data)} H1 bars, {data.index[0].date()} to {data.index[-1].date()}")
pair_results = {}
for phase in ["FULL", "TRAIN", "TEST"]:
t0 = time.time()
report, trade_log = run_test(pair, data, htf, phase)
elapsed = time.time() - t0
m = extract_metrics(report)
pair_results[phase] = m
print(f" {phase:>5}: {m['trades']:>3} trades | WR {m['wr']:>5.1f}% | "
f"PF {m['pf']:>5.2f} | PnL {m['pnl_pips']:>+8.1f}p | "
f"DD {m['dd']:>6.2f}% | AvgW {m['avg_win']:>5.1f} / "
f"AvgL {m['avg_loss']:>5.1f} | RR {m['rr']:>4.2f} | "
f"Exp {m['expectancy']:>+6.2f}p ({elapsed:.0f}s)")
if len(trade_log) > 0:
trade_log.to_csv(os.path.join(RESULTS_DIR, f"S7T_{pair}_{phase}_trades.csv"), index=False)
train_pf = pair_results.get("TRAIN", {}).get("pf", 0)
test_pf = pair_results.get("TEST", {}).get("pf", 0)
if train_pf > 0 and test_pf > 0:
print(f" Gen Ratio: {test_pf / train_pf:.2f}")
all_results[pair] = {"mode": "FULL_DATASET", **pair_results}
# ---- TEST 2: Other 5 pairs, 6 months ----
other_pairs = ["GBP_USD", "EUR_AUD", "GBP_AUD", "EUR_USD", "GBP_NZD"]
for pair in other_pairs:
print(f"\n{'#' * 70}")
print(f"# S7 Tight — {pair} (6 months)")
print(f"{'#' * 70}")
data = load_data(pair, last_n_months=6)
if data is None:
continue
htf = data.copy()
print(f" Data: {len(data)} H1 bars, {data.index[0].date()} to {data.index[-1].date()}")
pair_results = {}
for phase in ["FULL", "TRAIN", "TEST"]:
t0 = time.time()
report, trade_log = run_test(pair, data, htf, phase)
elapsed = time.time() - t0
m = extract_metrics(report)
pair_results[phase] = m
print(f" {phase:>5}: {m['trades']:>3} trades | WR {m['wr']:>5.1f}% | "
f"PF {m['pf']:>5.2f} | PnL {m['pnl_pips']:>+8.1f}p | "
f"DD {m['dd']:>6.2f}% | AvgW {m['avg_win']:>5.1f} / "
f"AvgL {m['avg_loss']:>5.1f} | RR {m['rr']:>4.2f} | "
f"Exp {m['expectancy']:>+6.2f}p ({elapsed:.0f}s)")
if len(trade_log) > 0:
trade_log.to_csv(os.path.join(RESULTS_DIR, f"S7T_{pair}_{phase}_trades.csv"), index=False)
train_pf = pair_results.get("TRAIN", {}).get("pf", 0)
test_pf = pair_results.get("TEST", {}).get("pf", 0)
if train_pf > 0 and test_pf > 0:
print(f" Gen Ratio: {test_pf / train_pf:.2f}")
all_results[pair] = {"mode": "6_MONTHS", **pair_results}
# ---- SUMMARY TABLE ----
print(f"\n{'=' * 90}")
print("S7 TIGHT (1.0 ATR SL) — ALL PAIRS SUMMARY")
print(f"{'=' * 90}")
print(f"{'Pair':<10} {'Mode':<12} {'Phase':<6} {'Trades':>6} {'WR%':>6} "
f"{'PF':>6} {'PnL(p)':>9} {'RR':>5} {'Exp(p)':>7} {'DD%':>7}")
print("-" * 90)
for pair, data in all_results.items():
mode = data["mode"]
for phase in ["FULL", "TEST"]:
m = data.get(phase, {})
if not m:
continue
print(f"{pair:<10} {mode:<12} {phase:<6} {m['trades']:>6} "
f"{m['wr']:>5.1f}% {m['pf']:>5.2f} {m['pnl_pips']:>+8.1f}p "
f"{m['rr']:>4.2f} {m['expectancy']:>+6.2f}p {m['dd']:>6.2f}%")
print()
# Save
out_path = os.path.join(RESULTS_DIR, "s7_expanded_results.json")
with open(out_path, "w") as f:
json.dump(all_results, f, indent=2)
print(f"Results saved: {out_path}")
src/run_s7_full_validation.py
+61
View File
@@ -0,0 +1,61 @@
"""S7 Tight — Full dataset validation for GBP_USD and EUR_USD."""
import os, sys, io, time
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
from src.strategies_pkg.s7_liquidity_sweep import S7_Liquidity_Sweep
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
os.makedirs(RESULTS_DIR, exist_ok=True)
TRAIN_RATIO = 0.70
for pair in ["GBP_USD", "EUR_USD"]:
print(f"\n{'#' * 70}")
print(f"# S7 Tight — {pair} FULL DATASET")
print(f"{'#' * 70}")
fp = os.path.join(PROCESSED_DIR, f"{pair}_H1.csv")
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
df = compute_all_indicators(df)
htf = df.copy()
print(f" Data: {len(df)} H1 bars, {df.index[0].date()} to {df.index[-1].date()}")
split_idx = int(len(df) * TRAIN_RATIO)
splits = {
"FULL": (df, htf),
"TRAIN": (df.iloc[:split_idx], htf.iloc[:split_idx]),
"TEST": (df.iloc[split_idx:], htf.iloc[split_idx:]),
}
for phase, (d, h) in splits.items():
t0 = time.time()
bt = Backtester(data=d, strategy=S7_Liquidity_Sweep(), pair=pair,
starting_equity=100_000.0, htf_data=h)
r = bt.run()
tl = bt.get_trade_log_df()
elapsed = time.time() - t0
n = r.get("total_trades", 0)
wr = r.get("win_rate_pct", 0)
pf = r.get("profit_factor", 0)
pnl = r.get("total_pnl_pips", 0)
pnl_d = r.get("total_pnl_dollars", 0)
dd = r.get("max_drawdown_pct", 0)
aw = r.get("avg_win_pips", 0)
al = r.get("avg_loss_pips", 0)
rr = aw / al if al > 0 else 0
exp = pnl / n if n > 0 else 0
print(f" {phase:>5}: {n:>3} trades | WR {wr:>5.1f}% | PF {pf:>5.2f} | "
f"PnL {pnl:>+8.1f}p (${pnl_d:>+10,.2f}) | DD {dd:>6.2f}% | "
f"AvgW {aw:>5.1f} / AvgL {al:>5.1f} | RR {rr:>4.2f} | "
f"Exp {exp:>+6.2f}p ({elapsed:.0f}s)")
if len(tl) > 0:
tl.to_csv(os.path.join(RESULTS_DIR, f"S7T_{pair}_FULL_{phase}_trades.csv"), index=False)
src/run_s9_additional_pairs.py
+155
View File
@@ -0,0 +1,155 @@
"""
S9 Full Dataset Test — Additional London-centric pairs.
Tests EUR_GBP, EUR_USD, GBP_NZD on the full 2021-2023 dataset.
(GBP_AUD and GBP_USD already tested previously.)
Uses FULL dataset with 70/30 train/test split.
"""
import os
import sys
import json
import time
import io
# Fix Windows encoding
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
# New pairs to test (London-centric: GBP or EUR)
PAIRS = ["EUR_GBP", "EUR_USD", "GBP_NZD"]
TRAIN_RATIO = 0.70
def load_data(pair, tf):
"""Load full dataset and compute indicators."""
fp = os.path.join(PROCESSED_DIR, f"{pair}_{tf}.csv")
if not os.path.exists(fp):
print(f" WARNING: {fp} not found, skipping")
return None
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
df = compute_all_indicators(df)
return df
def split_data(df, train_ratio=0.70):
split_idx = int(len(df) * train_ratio)
return df.iloc[:split_idx], df.iloc[split_idx:]
def run_pair(strat_cls, pair, data, htf_data, label, phase):
"""Run backtest on a single pair/phase."""
if phase == "TRAIN":
data, _ = split_data(data, TRAIN_RATIO)
htf_data, _ = split_data(htf_data, TRAIN_RATIO)
elif phase == "TEST":
_, data = split_data(data, TRAIN_RATIO)
_, htf_data = split_data(htf_data, TRAIN_RATIO)
strategy = strat_cls()
bt = Backtester(
data=data,
strategy=strategy,
pair=pair,
starting_equity=100_000.0,
htf_data=htf_data,
)
report = bt.run()
trade_log = bt.get_trade_log_df()
return report, trade_log
if __name__ == "__main__":
from src.strategies_pkg.s9_london_session import S9_London_Session
results = {}
for pair in PAIRS:
print(f"\n{'#' * 70}")
print(f"# S9 London Session — {pair} (Full Dataset)")
print(f"{'#' * 70}")
data = load_data(pair, "H1")
if data is None:
continue
htf_data = data.copy()
print(f" Data: {len(data)} H1 bars, {data.index[0].date()} to {data.index[-1].date()}")
pair_results = {}
for phase in ["FULL", "TRAIN", "TEST"]:
t0 = time.time()
report, trade_log = run_pair(S9_London_Session, pair, data, htf_data, "S9", phase)
elapsed = time.time() - t0
n = report.get("total_trades", 0)
wr = report.get("win_rate_pct", 0)
pf = report.get("profit_factor", 0)
pnl = report.get("total_pnl_pips", 0)
pnl_d = report.get("total_pnl_dollars", 0)
dd = report.get("max_drawdown_pct", 0)
avg_w = report.get("avg_win_pips", 0)
avg_l = report.get("avg_loss_pips", 0)
avg_rr = avg_w / avg_l if avg_l > 0 else 0
print(f"\n {phase}: {n} trades, WR {wr:.1f}%, PF {pf:.2f}, "
f"PnL {pnl:.1f}p (${pnl_d:,.2f}), DD {dd:.2f}%, "
f"AvgW {avg_w:.1f}p / AvgL {avg_l:.1f}p, RR {avg_rr:.2f} ({elapsed:.0f}s)")
pair_results[phase] = {
"total_trades": n, "win_rate": round(wr, 2),
"avg_rr": round(avg_rr, 2),
"expectancy": round(pnl / n, 2) if n > 0 else 0,
"profit_factor": round(pf, 2),
"max_dd": round(dd, 2),
"total_pnl_pips": round(pnl, 1),
"total_pnl_dollars": round(pnl_d, 2),
"avg_win": round(avg_w, 1), "avg_loss": round(avg_l, 1),
}
# Save trade log
if len(trade_log) > 0:
os.makedirs(RESULTS_DIR, exist_ok=True)
csv_path = os.path.join(RESULTS_DIR, f"S9_{pair}_{phase}_trades.csv")
trade_log.to_csv(csv_path, index=False)
# Generalization check
train_pf = pair_results.get("TRAIN", {}).get("profit_factor", 0)
test_pf = pair_results.get("TEST", {}).get("profit_factor", 0)
if train_pf > 0 and test_pf > 0:
gen = test_pf / train_pf
print(f"\n Generalization Ratio: {gen:.2f} (OOS PF / IS PF)")
results[pair] = pair_results
# Summary table
print(f"\n{'=' * 80}")
print("S9 ADDITIONAL PAIRS — SUMMARY")
print(f"{'=' * 80}")
print(f"{'Pair':<12} {'Phase':<8} {'Trades':>7} {'WR%':>7} {'PF':>7} {'PnL(p)':>10} {'DD%':>7} {'AvgW':>7} {'AvgL':>7} {'RR':>6}")
print("-" * 80)
for pair, phases in results.items():
for phase in ["FULL", "TRAIN", "TEST"]:
d = phases.get(phase, {})
print(f"{pair:<12} {phase:<8} {d.get('total_trades',0):>7} "
f"{d.get('win_rate',0):>6.1f}% {d.get('profit_factor',0):>6.2f} "
f"{d.get('total_pnl_pips',0):>9.1f}p {d.get('max_dd',0):>6.2f}% "
f"{d.get('avg_win',0):>6.1f} {d.get('avg_loss',0):>6.1f} "
f"{d.get('avg_rr',0):>5.2f}")
print()
# Save results
os.makedirs(RESULTS_DIR, exist_ok=True)
out_path = os.path.join(RESULTS_DIR, "s9_additional_pairs.json")
with open(out_path, "w") as f:
json.dump(results, f, indent=2)
print(f"Results saved: {out_path}")
src/run_s9_eurusd_light.py
+59
View File
@@ -0,0 +1,59 @@
"""S9 EUR_USD: Baseline vs Light Filter (RSI skip + TP1 1.5x only)."""
import os, sys, io, time
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
from src.strategies_pkg.s9_london_session import S9_London_Session
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
os.makedirs(RESULTS_DIR, exist_ok=True)
pair = "EUR_USD"
TRAIN_RATIO = 0.70
fp = os.path.join(PROCESSED_DIR, f"{pair}_H1.csv")
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
df = compute_all_indicators(df)
htf = df.copy()
print(f"Data: {len(df)} H1 bars, {df.index[0].date()} to {df.index[-1].date()}")
split_idx = int(len(df) * TRAIN_RATIO)
train_d, test_d = df.iloc[:split_idx], df.iloc[split_idx:]
train_h, test_h = htf.iloc[:split_idx], htf.iloc[split_idx:]
variants = {
"BASELINE": S9_London_Session(),
"LIGHT_FILTER": S9_London_Session(pair=pair, filtered=True),
}
print(f"\n{'Variant':<16} {'Phase':<6} {'Trades':>6} {'WR%':>6} {'PF':>6} "
f"{'PnL(p)':>9} {'RR':>5} {'Exp(p)':>7} {'DD%':>7}")
print("-" * 80)
for label, strat in variants.items():
for phase, d, h in [("FULL", df, htf), ("TRAIN", train_d, train_h), ("TEST", test_d, test_h)]:
s = S9_London_Session(pair=pair, filtered=(label != "BASELINE"))
bt = Backtester(data=d, strategy=s, pair=pair, starting_equity=100_000.0, htf_data=h)
r = bt.run()
tl = bt.get_trade_log_df()
n = r.get("total_trades", 0)
wr = r.get("win_rate_pct", 0)
pf = r.get("profit_factor", 0)
pnl = r.get("total_pnl_pips", 0)
dd = r.get("max_drawdown_pct", 0)
aw = r.get("avg_win_pips", 0)
al = r.get("avg_loss_pips", 0)
rr = aw / al if al > 0 else 0
exp = pnl / n if n > 0 else 0
print(f"{label:<16} {phase:<6} {n:>6} {wr:>5.1f}% {pf:>5.2f} "
f"{pnl:>+8.1f}p {rr:>4.2f} {exp:>+6.2f}p {dd:>6.2f}%")
if label == "LIGHT_FILTER" and len(tl) > 0:
tl.to_csv(os.path.join(RESULTS_DIR, f"S9FL_{pair}_{phase}_trades.csv"), index=False)
src/run_s9_filtered_comparison.py
+168
View File
@@ -0,0 +1,168 @@
"""
S9 Filtered vs Baseline comparison on EUR_USD and GBP_AUD.
Filters applied:
EUR_USD: ADX > 30, skip RSI 40-60, entry from Hour 8, TP1 = 1.5x range
GBP_AUD: ADX > 25, skip Friday, EMA50 distance > 40 pips
"""
import os, sys, io, json, time
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace')
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pandas as pd
from src.indicators.technical import compute_all_indicators
from src.backtester.engine import Backtester
from src.strategies_pkg.s9_london_session import S9_London_Session
PROCESSED_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "data", "processed")
RESULTS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "results", "phase1")
os.makedirs(RESULTS_DIR, exist_ok=True)
PAIRS = ["EUR_USD", "GBP_AUD"]
TRAIN_RATIO = 0.70
def load_data(pair):
fp = os.path.join(PROCESSED_DIR, f"{pair}_H1.csv")
df = pd.read_csv(fp, index_col=0, parse_dates=True)
df.index.name = "timestamp"
df = compute_all_indicators(df)
return df
def split_data(df):
split_idx = int(len(df) * TRAIN_RATIO)
return df.iloc[:split_idx], df.iloc[split_idx:]
def run_test(strategy, pair, data, htf_data, phase="FULL"):
if phase == "TRAIN":
data, _ = split_data(data)
htf_data, _ = split_data(htf_data)
elif phase == "TEST":
_, data = split_data(data)
_, htf_data = split_data(htf_data)
bt = Backtester(data=data, strategy=strategy, pair=pair,
starting_equity=100_000.0, htf_data=htf_data)
report = bt.run()
trade_log = bt.get_trade_log_df()
return report, trade_log
def extract_metrics(report, trade_log):
n = report.get("total_trades", 0)
wr = report.get("win_rate_pct", 0)
pf = report.get("profit_factor", 0)
pnl = report.get("total_pnl_pips", 0)
pnl_d = report.get("total_pnl_dollars", 0)
dd = report.get("max_drawdown_pct", 0)
avg_w = report.get("avg_win_pips", 0)
avg_l = report.get("avg_loss_pips", 0)
rr = avg_w / avg_l if avg_l > 0 else 0
exp = pnl / n if n > 0 else 0
return {
"trades": n, "wr": round(wr, 1), "pf": round(pf, 2),
"pnl_pips": round(pnl, 1), "pnl_usd": round(pnl_d, 2),
"dd": round(dd, 2), "avg_win": round(avg_w, 1),
"avg_loss": round(avg_l, 1), "rr": round(rr, 2),
"expectancy": round(exp, 2),
}
if __name__ == "__main__":
all_results = {}
for pair in PAIRS:
print(f"\n{'#' * 70}")
print(f"# {pair}")
print(f"{'#' * 70}")
data = load_data(pair)
htf_data = data.copy()
print(f" Data: {len(data)} H1 bars, {data.index[0].date()} to {data.index[-1].date()}")
pair_results = {}
for variant, filtered in [("BASELINE", False), ("FILTERED", True)]:
print(f"\n --- {variant} ---")
variant_results = {}
for phase in ["FULL", "TRAIN", "TEST"]:
strat = S9_London_Session(pair=pair, filtered=filtered)
t0 = time.time()
report, trade_log = run_test(strat, pair, data, htf_data, phase)
elapsed = time.time() - t0
m = extract_metrics(report, trade_log)
variant_results[phase] = m
print(f" {phase:>5}: {m['trades']:>3} trades | WR {m['wr']:>5.1f}% | "
f"PF {m['pf']:>5.2f} | PnL {m['pnl_pips']:>+8.1f}p | "
f"DD {m['dd']:>6.2f}% | AvgW {m['avg_win']:>5.1f} / "
f"AvgL {m['avg_loss']:>5.1f} | RR {m['rr']:>4.2f} | "
f"Exp {m['expectancy']:>+6.2f}p ({elapsed:.0f}s)")
# Save trade log for filtered version
if filtered and len(trade_log) > 0:
csv_path = os.path.join(RESULTS_DIR, f"S9F_{pair}_{phase}_trades.csv")
trade_log.to_csv(csv_path, index=False)
# Generalization ratio
train_pf = variant_results.get("TRAIN", {}).get("pf", 0)
test_pf = variant_results.get("TEST", {}).get("pf", 0)
if train_pf > 0 and test_pf > 0:
gen = test_pf / train_pf
print(f" Gen Ratio: {gen:.2f}")
pair_results[variant] = variant_results
# Delta summary
base_full = pair_results["BASELINE"]["FULL"]
filt_full = pair_results["FILTERED"]["FULL"]
base_test = pair_results["BASELINE"]["TEST"]
filt_test = pair_results["FILTERED"]["TEST"]
print(f"\n IMPROVEMENT (FULL):")
print(f" Trades: {base_full['trades']} -> {filt_full['trades']} "
f"({filt_full['trades'] - base_full['trades']:+d})")
print(f" WR: {base_full['wr']}% -> {filt_full['wr']}% "
f"({filt_full['wr'] - base_full['wr']:+.1f}%)")
print(f" PF: {base_full['pf']} -> {filt_full['pf']} "
f"({filt_full['pf'] - base_full['pf']:+.2f})")
print(f" PnL: {base_full['pnl_pips']:+.1f}p -> {filt_full['pnl_pips']:+.1f}p "
f"({filt_full['pnl_pips'] - base_full['pnl_pips']:+.1f}p)")
print(f" RR: {base_full['rr']} -> {filt_full['rr']} "
f"({filt_full['rr'] - base_full['rr']:+.2f})")
print(f"\n IMPROVEMENT (OOS TEST):")
print(f" Trades: {base_test['trades']} -> {filt_test['trades']}")
print(f" WR: {base_test['wr']}% -> {filt_test['wr']}%")
print(f" PF: {base_test['pf']} -> {filt_test['pf']}")
print(f" PnL: {base_test['pnl_pips']:+.1f}p -> {filt_test['pnl_pips']:+.1f}p")
all_results[pair] = pair_results
# Final comparison table
print(f"\n{'=' * 80}")
print("FINAL COMPARISON: BASELINE vs FILTERED")
print(f"{'=' * 80}")
print(f"{'Pair':<10} {'Variant':<10} {'Phase':<6} {'Trades':>7} {'WR%':>6} "
f"{'PF':>6} {'PnL(p)':>9} {'RR':>5} {'Exp(p)':>7}")
print("-" * 80)
for pair in PAIRS:
for variant in ["BASELINE", "FILTERED"]:
for phase in ["FULL", "TEST"]:
m = all_results[pair][variant][phase]
print(f"{pair:<10} {variant:<10} {phase:<6} {m['trades']:>7} "
f"{m['wr']:>5.1f}% {m['pf']:>5.2f} {m['pnl_pips']:>+8.1f}p "
f"{m['rr']:>4.2f} {m['expectancy']:>+6.2f}p")
if variant == "BASELINE":
print()
print("-" * 80)
# Save
out_path = os.path.join(RESULTS_DIR, "s9_filtered_comparison.json")
with open(out_path, "w") as f:
json.dump(all_results, f, indent=2)
print(f"\nResults saved: {out_path}")
src/strategies_pkg/__init__.py
+12
View File
@@ -5,6 +5,9 @@ from .s3_key_level_breakout import S3_KeyLevel_Breakout
from .s4_ema_ribbon import S4_EMA_Ribbon
from .s5_momentum_exhaustion import S5_Momentum_Exhaustion
from .s6_ema_bounce import S6_EMA_Bounce
from .s7_liquidity_sweep import S7_Liquidity_Sweep
from .s8_order_block import S8_Order_Block
from .s9_london_session import S9_London_Session
STRATEGIES = {
1: S1_MA_Breakout,
@@ -13,6 +16,9 @@ STRATEGIES = {
4: S4_EMA_Ribbon,
5: S5_Momentum_Exhaustion,
6: S6_EMA_Bounce,
7: S7_Liquidity_Sweep,
8: S8_Order_Block,
9: S9_London_Session,
}
# Which pairs each strategy trades
@@ -25,6 +31,9 @@ STRATEGY_PAIRS = {
4: ["GBP_AUD", "EUR_AUD", "GBP_JPY"],
5: ["GBP_AUD", "EUR_AUD", "GBP_JPY", "USD_JPY", "GBP_USD"],
6: ["GBP_AUD"], # Initial test — expand to EUR_AUD, GBP_USD if passing
7: ["GBP_USD", "GBP_JPY", "EUR_AUD"],
8: ["GBP_USD", "EUR_AUD", "GBP_JPY"],
9: ["GBP_USD", "EUR_AUD", "GBP_JPY"],
}
# Primary and filter timeframes
@@ -35,4 +44,7 @@ STRATEGY_TIMEFRAMES = {
4: {"primary": "M15", "filter": "H1"},
5: {"primary": "M15", "filter": "H1"},
6: {"primary": "M15", "filter": "H1"},
7: {"primary": "H1", "filter": None}, # H1 primary, internal HTF via htf_data
8: {"primary": "H1", "filter": None}, # H1 primary, internal HTF via htf_data
9: {"primary": "H1", "filter": None}, # H1 primary, internal HTF via htf_data
}
src/strategies_pkg/s7_liquidity_sweep.py
+275
View File
@@ -0,0 +1,275 @@
"""
Strategy S7: Liquidity Sweep Reversal.
Concept: Price sweeps past a significant swing high/low (triggering clustered
stop-loss orders), then reverses. The most empirically-supported SMC concept
(Osler 2005, NY Fed).
Entry conditions (ALL must be true):
1. Identify significant swing high/low (5-bar fractal) within last 100 bars
2. Price penetrates the swing level by 0.7-1.0 ATR (the sweep)
3. Price closes back inside the previous range (reversal candle)
4. OBV divergence: price makes new extreme but OBV doesn't confirm
(institutional absorption signal)
5. HTF (H1) trend alignment: only take sweeps in the direction of
the higher-timeframe trend (200 EMA bias)
6. Session filter: London/NY hours (08:00-17:00 UTC)
7. RSI < 35 (for longs) or > 65 (for shorts) as a filter
Exit:
- SL: 1.5 ATR beyond the sweep extreme
- TP1: 1.5 ATR from entry (close 50%)
- TP2: 3.0 ATR from entry (close 50%)
- Max hold: 40 bars (H1 = ~40 hours)
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S7_Liquidity_Sweep(BaseStrategy):
strategy_id = 7
name = "S7_Liquidity_Sweep"
# Tunable parameters
SWING_LOOKBACK = 5 # N-bar fractal for swing detection
SWING_HISTORY = 100 # How far back to search for swing levels
SWEEP_MIN_ATR = 0.7 # Minimum penetration for a sweep
SWEEP_MAX_ATR = 1.5 # Above this = genuine breakout, not a sweep
SL_ATR_MULT = 1.0 # SL beyond sweep extreme (tightened from 1.5)
TP1_ATR_MULT = 1.5 # First take-profit
TP2_ATR_MULT = 3.0 # Second take-profit
OBV_LOOKBACK = 20 # Lookback for OBV divergence detection
MAX_BARS = 40
def _find_swing_levels(self, data, idx):
"""Find significant swing highs and lows within lookback window."""
start = max(0, idx - self.SWING_HISTORY)
# Exclude the very recent bars (last 3) to avoid detecting current price action
end = idx - 2
if end - start < 20:
return [], []
swing_highs = []
swing_lows = []
lb = self.SWING_LOOKBACK
for i in range(start + lb, end - lb + 1):
# Swing high: highest high in [i-lb, i+lb]
window_highs = data["high"].iloc[i - lb:i + lb + 1]
if data["high"].iloc[i] == window_highs.max():
swing_highs.append((i, data["high"].iloc[i]))
# Swing low: lowest low in [i-lb, i+lb]
window_lows = data["low"].iloc[i - lb:i + lb + 1]
if data["low"].iloc[i] == window_lows.min():
swing_lows.append((i, data["low"].iloc[i]))
return swing_highs, swing_lows
def _find_equal_levels(self, levels, atr_val):
"""Find clusters of equal highs/lows (within 0.1 ATR) — highest probability targets."""
if len(levels) < 2:
return levels
tolerance = 0.1 * atr_val
clustered = []
used = set()
for i, (idx_i, price_i) in enumerate(levels):
if i in used:
continue
cluster = [(idx_i, price_i)]
used.add(i)
for j, (idx_j, price_j) in enumerate(levels):
if j in used:
continue
if abs(price_j - price_i) <= tolerance:
cluster.append((idx_j, price_j))
used.add(j)
if len(cluster) >= 2:
# Use the average price for the cluster, latest index
avg_price = np.mean([p for _, p in cluster])
latest_idx = max(idx for idx, _ in cluster)
clustered.append((latest_idx, avg_price))
else:
clustered.append((idx_i, price_i))
return clustered
def _check_obv_divergence(self, data, idx, direction):
"""Check for OBV divergence (institutional absorption signal)."""
lb = self.OBV_LOOKBACK
if idx < lb:
return False
window = data.iloc[idx - lb:idx + 1]
obv_vals = window.get("obv")
if obv_vals is None:
return False
if direction == "LONG":
# Bullish OBV divergence: price makes lower low but OBV makes higher low
price_lows = window["low"]
recent_low_pos = price_lows.values.argmin()
if recent_low_pos < lb - 5:
return False # Low isn't recent enough
# Find previous low in first half of window
first_half = price_lows.iloc[:lb // 2]
if len(first_half) < 3:
return False
prev_low_pos = first_half.values.argmin()
if (price_lows.iloc[recent_low_pos] < first_half.iloc[prev_low_pos] and
obv_vals.iloc[recent_low_pos] > obv_vals.iloc[prev_low_pos]):
return True
else:
# Bearish OBV divergence: price makes higher high but OBV makes lower high
price_highs = window["high"]
recent_high_pos = price_highs.values.argmax()
if recent_high_pos < lb - 5:
return False
first_half = price_highs.iloc[:lb // 2]
if len(first_half) < 3:
return False
prev_high_pos = first_half.values.argmax()
if (price_highs.iloc[recent_high_pos] > first_half.iloc[prev_high_pos] and
obv_vals.iloc[recent_high_pos] < obv_vals.iloc[prev_high_pos]):
return True
return False
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Session filter: 08:00-17:00 UTC
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
# HTF trend alignment
if htf_row is None:
return None
htf_ema200 = htf_row.get("ema_200", np.nan)
htf_close = htf_row.get("close", np.nan)
if np.isnan(htf_ema200) or np.isnan(htf_close):
return None
htf_bullish = htf_close > htf_ema200
htf_bearish = htf_close < htf_ema200
price = current["close"]
candle_high = current["high"]
candle_low = current["low"]
# RSI as confluence signal (not hard gate — lesson from S4)
rsi_val = current.get("rsi_14", 50)
if np.isnan(rsi_val):
rsi_val = 50
# Find swing levels
swing_highs, swing_lows = self._find_swing_levels(data, idx)
# ---- CHECK FOR BULLISH SWEEP (sweep below swing low, then reverse up) ----
if htf_bullish:
swing_lows = self._find_equal_levels(swing_lows, atr_val)
for sw_idx, sw_price in reversed(swing_lows): # Check most recent first
penetration = sw_price - candle_low
if penetration < self.SWEEP_MIN_ATR * atr_val:
continue
if penetration > self.SWEEP_MAX_ATR * atr_val:
continue
# Reversal confirmation: close back above the swing level
if price <= sw_price:
continue
# Strong close: in upper 40% of candle range
candle_range = candle_high - candle_low
if candle_range <= 0:
continue
if (price - candle_low) / candle_range < 0.4:
continue
# OBV divergence check (bonus confluence, not hard gate)
obv_div = self._check_obv_divergence(data, idx, "LONG")
confluence = 3 + (1 if obv_div else 0)
# RSI in oversold zone adds confluence (soft, not hard gate)
if rsi_val < 40:
confluence += 1
# Volume confirmation
vol = current.get("volume", 0)
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and vol > 1.5 * vol_avg:
confluence += 1
sweep_extreme = candle_low
sl = sweep_extreme - self.SL_ATR_MULT * atr_val
tp1 = price + self.TP1_ATR_MULT * atr_val
tp2 = price + self.TP2_ATR_MULT * atr_val
return {
"direction": "LONG",
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp2,
"confluence": confluence,
"entry_pattern": "liquidity_sweep_bullish",
"tp_splits": (0.50, 0.50, 0.0),
"trail_atr_mult": 1.5,
"max_bars": self.MAX_BARS,
}
# ---- CHECK FOR BEARISH SWEEP (sweep above swing high, then reverse down) ----
if htf_bearish:
swing_highs = self._find_equal_levels(swing_highs, atr_val)
for sw_idx, sw_price in reversed(swing_highs):
penetration = candle_high - sw_price
if penetration < self.SWEEP_MIN_ATR * atr_val:
continue
if penetration > self.SWEEP_MAX_ATR * atr_val:
continue
# Reversal: close back below the swing level
if price >= sw_price:
continue
# Strong close: in lower 40% of candle range
candle_range = candle_high - candle_low
if candle_range <= 0:
continue
if (candle_high - price) / candle_range < 0.4:
continue
obv_div = self._check_obv_divergence(data, idx, "SHORT")
confluence = 3 + (1 if obv_div else 0)
if rsi_val > 60:
confluence += 1
vol = current.get("volume", 0)
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and vol > 1.5 * vol_avg:
confluence += 1
sweep_extreme = candle_high
sl = sweep_extreme + self.SL_ATR_MULT * atr_val
tp1 = price - self.TP1_ATR_MULT * atr_val
tp2 = price - self.TP2_ATR_MULT * atr_val
return {
"direction": "SHORT",
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp2,
"confluence": confluence,
"entry_pattern": "liquidity_sweep_bearish",
"tp_splits": (0.50, 0.50, 0.0),
"trail_atr_mult": 1.5,
"max_bars": self.MAX_BARS,
}
return None
src/strategies_pkg/s8_order_block.py
+268
View File
@@ -0,0 +1,268 @@
"""
Strategy S8: Order Block Retest.
Concept: Price returns to the last opposing candle before a strong impulse
move (displacement), and bounces from it. Treated as classical supply/demand
zones with strict confluence filters (not mythical "institutional footprints").
Entry conditions (ALL must be true):
1. Detect a displacement candle: body >= 1.5 ATR (strong impulse)
2. Identify the order block: last opposing candle before displacement
3. Price returns to retest the OB zone (touches OB body range)
4. Rejection candle at OB: pin bar (wick >= 2x body) or engulfing pattern
5. At least 2 of 3 confluence factors:
a) FVG exists within the impulse move
b) OB is at a broken S/R level (structural confluence)
c) Volume declining on pullback into OB
6. HTF (H1) trend alignment via 200 EMA
7. Session filter: 08:00-17:00 UTC
Exit:
- SL: OB body low/high + 0.3 ATR buffer (NOT the full wick)
- TP1: 1.5 ATR from entry (close 50%)
- TP2: 3.0 ATR from entry (close 50%)
- Max hold: 40 bars
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S8_Order_Block(BaseStrategy):
strategy_id = 8
name = "S8_Order_Block"
# Tunable parameters
DISPLACEMENT_ATR = 1.5 # Min body size for displacement candle
DISPLACEMENT_VOL = 1.5 # Min volume ratio for displacement
OB_LOOKBACK = 50 # How far back to search for OBs
OB_RETEST_WINDOW = 30 # Max bars for price to retest OB after displacement
SL_ATR_BUFFER = 0.3 # Buffer beyond OB body for SL
TP1_ATR_MULT = 1.5
TP2_ATR_MULT = 3.0
MAX_BARS = 40
def _find_order_blocks(self, data, idx, atr_val):
"""
Find valid order blocks: last opposing candle before a displacement move.
Returns list of dicts: {direction, ob_idx, ob_body_high, ob_body_low,
ob_high, ob_low, displacement_idx, has_fvg}
"""
order_blocks = []
start = max(0, idx - self.OB_LOOKBACK)
for i in range(start + 1, idx - 2):
curr = data.iloc[i]
body = abs(curr["close"] - curr["open"])
# Is this a displacement candle? (body >= 1.5 ATR)
bar_atr = curr.get("atr_14", atr_val)
if np.isnan(bar_atr) or bar_atr <= 0:
bar_atr = atr_val
if body < self.DISPLACEMENT_ATR * bar_atr:
continue
# Volume confirmation for displacement
vol = curr.get("volume", 0)
vol_avg = data["volume"].iloc[max(0, i - 20):i].mean()
if vol_avg > 0 and vol < self.DISPLACEMENT_VOL * vol_avg:
continue
is_bullish_displacement = curr["close"] > curr["open"]
is_bearish_displacement = curr["close"] < curr["open"]
if not (is_bullish_displacement or is_bearish_displacement):
continue
# Find the order block: last OPPOSING candle before displacement
ob_idx = None
for j in range(i - 1, max(start, i - 10) - 1, -1):
ob_candle = data.iloc[j]
ob_bullish = ob_candle["close"] > ob_candle["open"]
ob_bearish = ob_candle["close"] < ob_candle["open"]
if is_bullish_displacement and ob_bearish:
ob_idx = j
break
elif is_bearish_displacement and ob_bullish:
ob_idx = j
break
if ob_idx is None:
continue
ob = data.iloc[ob_idx]
ob_body_high = max(ob["open"], ob["close"])
ob_body_low = min(ob["open"], ob["close"])
# Check for FVG in the impulse move
has_fvg = False
if i >= 2:
candle_before = data.iloc[i - 1]
candle_after_idx = min(i + 1, len(data) - 1)
candle_after = data.iloc[candle_after_idx]
if is_bullish_displacement:
# Bullish FVG: candle[i-1].high < candle[i+1].low
if candle_before["high"] < candle_after["low"]:
has_fvg = True
else:
# Bearish FVG: candle[i-1].low > candle[i+1].high
if candle_before["low"] > candle_after["high"]:
has_fvg = True
direction = "LONG" if is_bullish_displacement else "SHORT"
order_blocks.append({
"direction": direction,
"ob_idx": ob_idx,
"ob_body_high": ob_body_high,
"ob_body_low": ob_body_low,
"ob_high": ob["high"],
"ob_low": ob["low"],
"displacement_idx": i,
"has_fvg": has_fvg,
})
return order_blocks
def _is_rejection_candle(self, candle, direction):
"""Check if candle shows rejection (pin bar or engulfing-like)."""
body = abs(candle["close"] - candle["open"])
full_range = candle["high"] - candle["low"]
if full_range <= 0:
return False
if direction == "LONG":
lower_wick = min(candle["open"], candle["close"]) - candle["low"]
# Pin bar: lower wick >= 2x body, close in upper 40%
if lower_wick >= 2 * body and (candle["close"] - candle["low"]) / full_range >= 0.6:
return True
# Bullish candle with strong close
if candle["close"] > candle["open"] and body / full_range >= 0.5:
return True
else:
upper_wick = candle["high"] - max(candle["open"], candle["close"])
# Pin bar: upper wick >= 2x body, close in lower 40%
if upper_wick >= 2 * body and (candle["high"] - candle["close"]) / full_range >= 0.6:
return True
# Bearish candle with strong close
if candle["close"] < candle["open"] and body / full_range >= 0.5:
return True
return False
def _check_volume_declining(self, data, displacement_idx, idx):
"""Check if volume is declining on the pullback to OB."""
if idx <= displacement_idx + 2:
return False
displacement_vol = data["volume"].iloc[displacement_idx]
pullback_vol = data["volume"].iloc[displacement_idx + 1:idx + 1].mean()
return pullback_vol < 0.8 * displacement_vol
def _is_at_broken_sr(self, data, idx, ob_price, atr_val):
"""Check if OB is at a level where prior S/R was broken (structural confluence)."""
# Look for swing highs/lows near the OB price that were broken
tolerance = 0.5 * atr_val
lookback_start = max(0, idx - 200)
for i in range(lookback_start, idx - 20):
bar = data.iloc[i]
is_sh = bar.get("is_swing_high", False)
is_sl_point = bar.get("is_swing_low", False)
if is_sh and abs(bar["high"] - ob_price) < tolerance:
return True
if is_sl_point and abs(bar["low"] - ob_price) < tolerance:
return True
return False
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Session filter
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
# HTF trend alignment
if htf_row is None:
return None
htf_ema200 = htf_row.get("ema_200", np.nan)
htf_close = htf_row.get("close", np.nan)
if np.isnan(htf_ema200) or np.isnan(htf_close):
return None
price = current["close"]
# Find order blocks
order_blocks = self._find_order_blocks(data, idx, atr_val)
for ob in order_blocks:
# Only trade OBs aligned with HTF trend
if ob["direction"] == "LONG" and htf_close < htf_ema200:
continue
if ob["direction"] == "SHORT" and htf_close > htf_ema200:
continue
# Check if price is retesting the OB zone
# For LONG: price should be in or near the OB body zone (pullback down into it)
if ob["direction"] == "LONG":
if not (current["low"] <= ob["ob_body_high"] and price >= ob["ob_body_low"]):
continue
else:
if not (current["high"] >= ob["ob_body_low"] and price <= ob["ob_body_high"]):
continue
# Check OB is not too old (retest within window)
bars_since = idx - ob["displacement_idx"]
if bars_since > self.OB_RETEST_WINDOW or bars_since < 3:
continue
# Rejection candle check
if not self._is_rejection_candle(current, ob["direction"]):
continue
# Confluence scoring (need 2 of 3)
confluence_count = 0
if ob["has_fvg"]:
confluence_count += 1
if self._check_volume_declining(data, ob["displacement_idx"], idx):
confluence_count += 1
ob_mid = (ob["ob_body_high"] + ob["ob_body_low"]) / 2
if self._is_at_broken_sr(data, idx, ob_mid, atr_val):
confluence_count += 1
if confluence_count < 2:
continue
# Build exit levels using OB BODY (not wick) + buffer
if ob["direction"] == "LONG":
sl = ob["ob_body_low"] - self.SL_ATR_BUFFER * atr_val
tp1 = price + self.TP1_ATR_MULT * atr_val
tp2 = price + self.TP2_ATR_MULT * atr_val
else:
sl = ob["ob_body_high"] + self.SL_ATR_BUFFER * atr_val
tp1 = price - self.TP1_ATR_MULT * atr_val
tp2 = price - self.TP2_ATR_MULT * atr_val
return {
"direction": ob["direction"],
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp2,
"confluence": min(confluence_count + 2, 5),
"entry_pattern": f"order_block_retest_{ob['direction'].lower()}",
"tp_splits": (0.50, 0.50, 0.0),
"trail_atr_mult": 1.5,
"max_bars": self.MAX_BARS,
}
return None
src/strategies_pkg/s9_london_session.py
+270
View File
@@ -0,0 +1,270 @@
"""
Strategy S9: London Session Gap (Asian Range Breakout).
Concept: Price breaks out of the Asian session range at London open, driven
by institutional order flow from European/UK desks. Session-based volatility
patterns are among the most well-documented phenomena in FX
(Andersen & Bollerslev 1997, BIS data).
Entry conditions (ALL must be true):
1. Asian range defined: 00:00-07:00 UTC high/low
2. Asian range not too wide (< 1.5 ATR H1 and < pair-specific cap)
3. Price breaks above Asian high (LONG) or below Asian low (SHORT)
with a candle CLOSE beyond the level
4. Volume > 3.0x Asian session average (first London candle almost always
shows 2x, so 3x filters for meaningful surges)
5. ADX > 20 (some trending context)
6. Trade window: 07:00-10:00 UTC (London kill zone)
Exit:
- SL: Opposite side of Asian range, capped at 1.5 ATR(H1) or pip limit
- TP1: Asian range width as measured-move target (close 50%)
- TP2: 2.0x Asian range width (close 50%)
- Time exit: 17:00 UTC (captures full London-NY overlap)
- Max hold: 40 bars (H1)
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S9_London_Session(BaseStrategy):
strategy_id = 9
name = "S9_London_Session"
# Asian range window (UTC hours)
ASIAN_START_HOUR = 0
ASIAN_END_HOUR = 7
# Entry window (UTC hours)
ENTRY_START_HOUR = 7
ENTRY_END_HOUR = 10
# Exit time (UTC hour) — captures full London-NY overlap
TIME_EXIT_HOUR = 17
# Volume threshold (relaxed from 3.0 for H1 — Asian H1 bars aren't dramatically
# lower volume than London H1 bars the way M15 bars would be)
VOLUME_MULT = 1.5
# Max Asian range (in pips) per pair category
MAX_RANGE_PIPS = {
"EUR_USD": 60, "GBP_USD": 80, "EUR_AUD": 80,
"GBP_AUD": 100, "GBP_JPY": 100, "USD_JPY": 60,
"EUR_CAD": 80, "GBP_CAD": 100, "EUR_GBP": 50,
}
# SL cap in ATR (widened from 1.5 — was filtering out most days)
SL_ATR_CAP = 2.5
MAX_BARS = 40
# Per-pair filter overrides (set via constructor with pair= and filtered=True)
# Each key maps to a dict of: min_adx, rsi_neutral_skip, skip_friday,
# entry_start_hour, tp1_mult, min_ema50_dist_pips
PAIR_FILTERS = {
"EUR_USD": {
"tp1_mult": 1.5, # TP1 = 1.5x Asian range (was 1.0x)
# NOTE: RSI filter and ADX hard gate tested but overfit — dropped
},
"GBP_AUD": {
"min_adx": 25, # require ADX > 25
"skip_friday": True, # drop Friday trades
"min_ema50_dist_pips": 40, # require 40+ pips from EMA50
},
}
def __init__(self, pair=None, filtered=False):
super().__init__()
self._asian_range_cache = {} # date -> (high, low, avg_vol)
self._pair = pair
self._filtered = filtered
self._pair_cfg = {}
if filtered and pair and pair in self.PAIR_FILTERS:
self._pair_cfg = self.PAIR_FILTERS[pair]
def _get_pip_size(self, pair):
if "JPY" in pair:
return 0.01
return 0.0001
def _compute_asian_range(self, data, idx):
"""Compute Asian session range for the current day."""
current_time = data.index[idx]
current_date = current_time.date()
if current_date in self._asian_range_cache:
return self._asian_range_cache[current_date]
# Find Asian session bars for today (00:00-07:00 UTC)
asian_bars = []
for i in range(max(0, idx - 50), idx + 1):
bar_time = data.index[i]
if bar_time.date() != current_date:
continue
bar_hour = bar_time.hour
if self.ASIAN_START_HOUR <= bar_hour < self.ASIAN_END_HOUR:
asian_bars.append(i)
if len(asian_bars) < 3:
return None
asian_data = data.iloc[asian_bars]
asian_high = asian_data["high"].max()
asian_low = asian_data["low"].min()
asian_avg_vol = asian_data["volume"].mean()
result = (asian_high, asian_low, asian_avg_vol)
self._asian_range_cache[current_date] = result
return result
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Entry window (per-pair override for start hour)
hour = current.name.hour if hasattr(current.name, 'hour') else 0
start_hour = self._pair_cfg.get("entry_start_hour", self.ENTRY_START_HOUR)
if hour < start_hour or hour >= self.ENTRY_END_HOUR:
return None
# Friday filter (GBP_AUD: Friday position squaring kills breakouts)
if self._pair_cfg.get("skip_friday", False):
dow = current.name.dayofweek if hasattr(current.name, 'dayofweek') else 0
if dow == 4: # Friday
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
# Get HTF ATR for SL capping
htf_atr = atr_val
if htf_row is not None:
htf_atr_val = htf_row.get("atr_14", np.nan)
if not np.isnan(htf_atr_val) and htf_atr_val > 0:
htf_atr = htf_atr_val
# Compute Asian range
asian = self._compute_asian_range(data, idx)
if asian is None:
return None
asian_high, asian_low, asian_avg_vol = asian
asian_range = asian_high - asian_low
if asian_range <= 0:
return None
# Check Asian range not too wide
pair = ""
# Try to infer pair from strategy context; use default cap
max_range_pips = 80 # default
pip_size = self._get_pip_size("GBP_JPY" if atr_val > 0.005 else "EUR_USD")
range_pips = asian_range / pip_size
# Cap: skip if Asian range > 1.5 ATR(H1)
if asian_range > self.SL_ATR_CAP * htf_atr:
return None
price = current["close"]
# ADX filter — hard gate when filtered, soft confluence otherwise
adx_val = current.get("adx_14", 0)
if np.isnan(adx_val):
adx_val = 0
min_adx = self._pair_cfg.get("min_adx", 0)
if min_adx > 0 and adx_val < min_adx:
return None
adx_strong = adx_val > 20
# RSI neutral zone filter (EUR_USD: skip RSI 40-60 — no directional momentum)
if self._pair_cfg.get("rsi_neutral_skip", False):
rsi_val = current.get("rsi_14", 50)
if not np.isnan(rsi_val) and 40 <= rsi_val <= 60:
return None
# EMA50 distance filter (GBP_AUD: close-to-EMA trades underperform)
min_ema_dist = self._pair_cfg.get("min_ema50_dist_pips", 0)
if min_ema_dist > 0:
ema50 = current.get("ema_50", np.nan)
if not np.isnan(ema50) and ema50 > 0:
pip_sz = self._get_pip_size(self._pair or "EUR_USD")
dist_pips = abs(price - ema50) / pip_sz
if dist_pips < min_ema_dist:
return None
# Volume check: current volume > 1.5x Asian average
vol = current.get("volume", 0)
if asian_avg_vol <= 0 or vol < self.VOLUME_MULT * asian_avg_vol:
return None
# Direction: breakout above or below Asian range
direction = None
if price > asian_high and current["close"] > asian_high:
direction = "LONG"
elif price < asian_low and current["close"] < asian_low:
direction = "SHORT"
if direction is None:
return None
# HTF trend alignment (soft: adds confluence but doesn't block)
htf_aligned = False
if htf_row is not None:
htf_ema200 = htf_row.get("ema_200", np.nan)
htf_close = htf_row.get("close", np.nan)
if not np.isnan(htf_ema200) and not np.isnan(htf_close):
if direction == "LONG" and htf_close > htf_ema200:
htf_aligned = True
elif direction == "SHORT" and htf_close < htf_ema200:
htf_aligned = True
confluence = 3 + (1 if htf_aligned else 0) + (1 if adx_strong else 0)
# SL: opposite side of Asian range, capped
tp1_mult = self._pair_cfg.get("tp1_mult", 1.0)
if direction == "LONG":
raw_sl = asian_low
sl_distance = price - raw_sl
max_sl_distance = self.SL_ATR_CAP * htf_atr
if sl_distance > max_sl_distance:
raw_sl = price - max_sl_distance
sl = raw_sl
tp1 = price + tp1_mult * asian_range
tp2 = price + 2.0 * asian_range
else:
raw_sl = asian_high
sl_distance = raw_sl - price
max_sl_distance = self.SL_ATR_CAP * htf_atr
if sl_distance > max_sl_distance:
raw_sl = price + max_sl_distance
sl = raw_sl
tp1 = price - tp1_mult * asian_range
tp2 = price - 2.0 * asian_range
# Calculate max bars until 17:00 UTC time exit
# On H1: roughly 17 - current_hour bars; on M15: (17-hour)*4
# Use generic max_bars as fallback
hours_remaining = self.TIME_EXIT_HOUR - hour
if hours_remaining <= 0:
return None
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp2,
"confluence": confluence,
"entry_pattern": f"london_breakout_{direction.lower()}",
"tp_splits": (0.50, 0.50, 0.0),
"trail_atr_mult": 1.5,
"max_bars": self.MAX_BARS,
}