session-scalping

> **Skill:** Session Scalping | **Domain:** trading | **Category:** strategy | **Level:** intermediate > **Tags:** `trading`, `strategy`, `scalping`, `orb`, `session`, `ny-open` ## Asian Session Scalper # Asian Session Scalper ```python import pandas as pd, numpy as np class AsianSessionScalper: @staticmethod def range_fade(df: pd.DataFrame) -> dict: """Fade the range during Tokyo session — buy lows, sell highs of the range.""" df = df.copy() df["hour"] = df.index.hour asian = df[(df["hour"] >= 0) & (df["hour"] < 7)] if len(asian) < 10: return {"error": "Insufficient Asian data"} range_high = asian["high"].rolling(20).max().iloc[-1] range_low = asian["low"].rolling(20).min().iloc[-1] mid = (range_high + range_low) / 2 current = df.iloc[-1]["close"] atr = (asian["high"] - asian["low"]).mean() return { "strategy": "asian_range_fade", "range_high": round(range_high, 5), "range_low": round(range_low, 5), "midpoint": round(mid, 5), "signal": "BUY (near range low)" if current < range_low + atr * 0.3 else "SELL (near range high)" if current > range_high - atr * 0.3 else "WAIT (mid-range)", "stop_pips": round(atr * 10000 * 1.5, 1), "target_pips": round(atr * 10000 * 1.0, 1), "best_pairs": ["USDJPY", "EURJPY", "AUDJPY", "AUDNZD"], "avoid": ["GBPUSD", "EURUSD (low liquidity in Asia)"], } ``` --- ## Session Breakout Strategies # Session Breakout Strategies ```python import pandas as pd, numpy as np from datetime import time class SessionBreakoutStrategies: @staticmethod def asian_range_breakout(df: pd.DataFrame) -> dict: """Trade the breakout of the Asian session range during London open.""" df = df.copy() df["hour"] = df.index.hour asian = df[(df["hour"] >= 0) & (df["hour"] < 7)] if asian.empty: return {"error": "No Asian session data"} asian_high = asian["high"].max() asian_low = asian["low"].min() asian_range = asian_high - asian_low current = df.iloc[-1] return { "strategy": "asian_range_breakout", "asian_high": round(asian_high, 5), "asian_low": round(asian_low, 5), "range_pips": round(asian_range * 10000, 1), "buy_trigger": round(asian_high, 5), "sell_trigger": round(asian_low, 5), "buy_sl": round(asian_low, 5), "sell_sl": round(asian_high, 5), "buy_tp": round(asian_high + asian_range, 5), "sell_tp": round(asian_low - asian_range, 5), "broken_up": current["close"] > asian_high, "broken_down": current["close"] < asian_low, "timing": "Place pending orders at 07:00 UTC (London open)", "cancel_by": "12:00 UTC if not triggered", "best_pairs": ["GBPUSD", "EURUSD", "EURGBP"], } @staticmethod def london_breakout(df: pd.DataFrame) -> dict: """Trade first directional move of London session.""" df = df.copy() df["hour"] = df.index.hour first_hour = df[(df["hour"] >= 7) & (df["hour"] < 8)] if first_hour.empty: return {"error": "No London first hour data"} fh_high = first_hour["high"].max() fh_low = first_hour["low"].min() fh_range = fh_high - fh_low current = df.iloc[-1] return { "strategy": "london_breakout", "first_hour_high": round(fh_high, 5), "first_hour_low": round(fh_low, 5), "buy_trigger": round(fh_high, 5), "sell_trigger": round(fh_low, 5), "target": round(fh_range * 1.5, 5), "stop": round(fh_range * 0.75, 5), "broken_up": current["close"] > fh_high, "broken_down": current["close"] < fh_low, "timing": "08:00-10:00 UTC", "best_days": "Tuesday, Wednesday, Thursday", } @staticmethod def ny_session_reversal(df: pd.DataFrame) -> dict: """NY session often reverses the London move. Fade London direction after NY open.""" df = df.copy() df["hour"] = df.index.hour london = df[(df["hour"] >= 7) & (df["hour"] < 13)] if london.empty: return {"error": "No London data"} london_direction = "UP" if london["close"].iloc[-1] > london["open"].iloc[0] else "DOWN" london_move = abs(london["close"].iloc[-1] - london["open"].iloc[0]) atr = (df["high"] - df["low"]).rolling(14).mean().iloc[-1] extended = london_move > 1.5 * atr return { "strategy": "ny_reversal", "london_direction": london_direction, "london_move_pips": round(london_move * 10000, 1), "extended": extended, "signal": f"FADE {london_direction} — sell if London went UP, buy if DOWN" if extended else "WAIT — London move not extended enough", "timing": "13:30-15:00 UTC (after NY data releases)", "confirmation": "Wait for rejection candle at London extreme before fading", } ``` --- ## Scalping Framework # Scalping Framework ## CRITICAL: Only scalp during HIGH LIQUIDITY sessions (London/NY overlap). Spread must be < 1.5 pips. ```python import pandas as pd import numpy as np class ScalpingFramework: @staticmethod def spread_check(current_spread_pips: float, avg_spread: float) -> dict: """Pre-scalp spread validation — never scalp with wide spreads.""" ratio = current_spread_pips / max(avg_spread, 0.1) return { "current_spread": current_spread_pips, "avg_spread": avg_spread, "spread_ratio": round(ratio, 2), "can_scalp": current_spread_pips < 1.5 and ratio < 1.5, "warning": "SPREAD TOO WIDE — do not scalp" if current_spread_pips > 2.0 else None, } @staticmethod def momentum_burst(df: pd.DataFrame, lookback: int = 5, threshold_mult: float = 2.0) -> dict: """Detect sudden momentum bursts for scalp entries.""" close = df["close"] returns = close.pct_change() avg_move = returns.rolling(50).std() burst = returns.abs() > threshold_mult * avg_move direction = np.where(returns > 0, "long", "short") current_burst = burst.iloc[-1] return { "strategy": "momentum_burst_scalp", "burst_detected": bool(current_burst), "direction": direction[-1] if current_burst else "none", "magnitude": round(abs(returns.iloc[-1]) / avg_move.iloc[-1], 1) if avg_move.iloc[-1] > 0 else 0, "entry": round(close.iloc[-1], 5), "target_pips": round(avg_move.iloc[-1] * 10000 * 1.5, 1), "stop_pips": round(avg_move.iloc[-1] * 10000 * 1.0, 1), "max_hold_bars": 10, } @staticmethod def ema_cross_scalp(df: pd.DataFrame, fast: int = 5, slow: int = 13) -> dict: """Ultra-fast EMA crossover for M1/M5 scalping.""" close = df["close"] ema_fast = close.ewm(span=fast).mean() ema_slow = close.ewm(span=slow).mean() cross_up = (ema_fast.iloc[-1] > ema_slow.iloc[-1]) and (ema_fast.iloc[-2] <= ema_slow.iloc[-2]) cross_down = (ema_fast.iloc[-1] < ema_slow.iloc[-1]) and (ema_fast.iloc[-2] >= ema_slow.iloc[-2]) return { "strategy": "ema_cross_scalp", "fast_ema": round(ema_fast.iloc[-1], 5), "slow_ema": round(ema_slow.iloc[-1], 5), "cross_up": cross_up, "cross_down": cross_down, "signal": "LONG" if cross_up else "SHORT" if cross_down else "WAIT", "hold_max_bars": 15, } @staticmethod def scalp_rules() -> dict: return { "max_hold_time": "15-30 minutes (M1) or 1-2 hours (M5)", "max_risk_per_scalp": "0.5% of account (half normal risk)", "min_rr": "1:1 minimum (1:1.5 preferred)", "session": "London/NY overlap ONLY (13:00-16:00 UTC)", "spread_max": "1.5 pips (ideally < 1.0)", "pairs": "EURUSD, GBPUSD, USDJPY only (tightest spreads)", "stop_after": "3 consecutive losses — take a break", } ``` --- ## Breakout Strategy Engine # Breakout Strategy Engine ## Pre-Built Breakout Strategies with Confirmation Filters ```python import pandas as pd import numpy as np from dataclasses import dataclass from typing import Optional @dataclass class BreakoutSignal: symbol: str direction: str # "long" or "short" entry: float stop_loss: float target: float strategy: str confirmation: list[str] strength: float # 0-1 class BreakoutEngine: # ═══════════════════════════════════════ # 1. BOLLINGER SQUEEZE BREAKOUT # ═══════════════════════════════════════ @staticmethod def bollinger_squeeze(df: pd.DataFrame, bb_period: int = 20, kc_period: int = 20, kc_mult: float = 1.5) -> dict: """Bollinger inside Keltner Channel = squeeze. Breakout when squeeze releases.""" close = df["close"] bb_mid = close.rolling(bb_period).mean() bb_std = close.rolling(bb_period).std() bb_upper = bb_mid + 2 * bb_std bb_lower = bb_mid - 2 * bb_std atr = ((df["high"] - df["low"]).rolling(kc_period).mean()) kc_upper = bb_mid + kc_mult * atr kc_lower = bb_mid - kc_mult * atr squeeze_on = (bb_lower > kc_lower) & (bb_upper < kc_upper) squeeze_off = ~squeeze_on # Squeeze just released squeeze_fire = squeeze_off & squeeze_on.shift(1) # Direction from momentum momentum = close - close.rolling(bb_period).mean() direction = np.where(momentum > 0, "long", "short") df_out = df.copy() df_out["squeeze_on"] = squeeze_on df_out["squeeze_fire"] = squeeze_fire df_out["direction"] = direction df_out["bb_width"] = (bb_upper - bb_lower) / bb_mid * 100 current = df_out.iloc[-1] return { "strategy": "bollinger_squeeze", "squeeze_active": bool(current["squeeze_on"]), "squeeze_firing": bool(current["squeeze_fire"]), "direction": current["direction"], "bb_width": round(current["bb_width"], 3),