mahmoud20138 Session Profiler
Overview
Statistical profiles of major trading sessions with historical tendencies, volatility patterns, and pair-specific behavior. Use for optimal entry timing and understanding session-driven flows.
1. Session Definitions
from datetime import datetime, time, timedelta
from typing import Optional
import pandas as pd
import numpy as np
SESSIONS = {
"tokyo": {
"name": "Tokyo / Asian",
"open_utc": time(0, 0), "close_utc": time(9, 0),
"peak_utc": (time(1, 0), time(6, 0)),
"primary_pairs": ["USDJPY", "EURJPY", "GBPJPY", "AUDJPY", "AUDUSD", "NZDUSD"],
"characteristics": "Low volatility, range-bound. Good for range strategies.",
},
"london": {
"name": "London / European",
"open_utc": time(7, 0), "close_utc": time(16, 0),
"peak_utc": (time(7, 0), time(11, 0)),
"primary_pairs": ["EURUSD", "GBPUSD", "EURGBP", "EURJPY", "GBPJPY", "USDCHF"],
"characteristics": "Highest volume session. Breakouts from Asian range. Most trending moves.",
},
"new_york": {
"name": "New York / US",
"open_utc": time(13, 0), "close_utc": time(22, 0),
"peak_utc": (time(13, 0), time(17, 0)),
"primary_pairs": ["EURUSD", "GBPUSD", "USDJPY", "USDCAD", "USDCHF", "XAUUSD"],
"characteristics": "Second highest volume. Major economic releases. Often reverses London moves.",
},
"london_ny_overlap": {
"name": "London-NY Overlap",
"open_utc": time(13, 0), "close_utc": time(16, 0),
"peak_utc": (time(13, 0), time(16, 0)),
"primary_pairs": ["ALL"],
"characteristics": "Highest volatility window of the day. Maximum liquidity.",
},
}
# ICT Killzones
ICT_KILLZONES = {
"asian_kz": {"start": time(0, 0), "end": time(4, 0), "name": "Asian Killzone"},
"london_kz": {"start": time(7, 0), "end": time(10, 0), "name": "London Open Killzone"},
"ny_kz": {"start": time(13, 0), "end": time(16, 0), "name": "NY Open Killzone"},
"london_close": {"start": time(15, 0), "end": time(16, 0), "name": "London Close Killzone"},
}
def current_session(now: datetime = None) -> dict:
"""Determine current active session(s)."""
now = now or datetime.utcnow()
t = now.time()
active = []
for key, session in SESSIONS.items():
if session["open_utc"] <= t <= session["close_utc"]:
in_peak = session["peak_utc"][0] <= t <= session["peak_utc"][1]
active.append({"session": key, "name": session["name"], "in_peak": in_peak})
killzones = []
for key, kz in ICT_KILLZONES.items():
if kz["start"] <= t <= kz["end"]:
killzones.append(kz["name"])
return {"active_sessions": active, "killzones": killzones, "time_utc": now.strftime("%H:%M")}
2. Session Statistics Engine
class SessionProfiler:
"""Compute statistical profiles per session from historical data."""
@staticmethod
def session_stats(df: pd.DataFrame, pair: str = "") -> dict:
"""Compute per-session statistics from OHLCV data."""
df = df.copy()
df["hour"] = df.index.hour
df["session"] = df["hour"].apply(lambda h:
"tokyo" if 0 <= h < 7 else
"london" if 7 <= h < 13 else
"ny_overlap" if 13 <= h < 16 else
"ny_late" if 16 <= h < 22 else "off_hours"
)
df["range"] = df["high"] - df["low"]
df["body"] = abs(df["close"] - df["open"])
df["direction"] = np.where(df["close"] > df["open"], 1, -1)
stats = {}
for session in ["tokyo", "london", "ny_overlap", "ny_late"]:
s = df[df["session"] == session]
if s.empty:
continue
stats[session] = {
"avg_range_pips": round(s["range"].mean() * 10000, 1),
"max_range_pips": round(s["range"].max() * 10000, 1),
"avg_body_pips": round(s["body"].mean() * 10000, 1),
"bullish_pct": round((s["direction"] == 1).mean() * 100, 1),
"bearish_pct": round((s["direction"] == -1).mean() * 100, 1),
"bars_analyzed": len(s),
"avg_volume": round(s["volume"].mean(), 0) if "volume" in s.columns else 0,
}
return {"pair": pair, "session_stats": stats}
@staticmethod
def session_open_patterns(df: pd.DataFrame) -> dict:
"""Analyze behavior around session opens."""
df = df.copy()
df["hour"] = df.index.hour
patterns = {}
for session, open_hour in [("london", 7), ("new_york", 13)]:
opens = df[df["hour"] == open_hour]
if opens.empty:
continue
# First hour direction
first_hour_up = (opens["close"] > opens["open"]).mean()
# Continuation: does the first hour direction hold?
patterns[session] = {
"first_bar_bullish_pct": round(first_hour_up * 100, 1),
"avg_first_bar_range": round((opens["high"] - opens["low"]).mean() * 10000, 1),
"note": f"{session.title()} open tends {'bullish' if first_hour_up > 0.55 else 'bearish' if first_hour_up < 0.45 else 'neutral'}",
}
return patterns
@staticmethod
def day_of_week_profile(df: pd.DataFrame) -> pd.DataFrame:
"""Statistical profile by day of week."""
df = df.copy()
df["dow"] = df.index.day_name()
df["range"] = df["high"] - df["low"]
return df.groupby("dow").agg(
avg_range=("range", lambda x: round(x.mean() * 10000, 1)),
bullish_pct=("close", lambda x: round((x > x.shift(1)).mean() * 100, 1)),
avg_volume=("volume", "mean"),
).reindex(["Monday", "Tuesday", "Wednesday", "Thursday", "Friday"])
@staticmethod
def hourly_volatility_profile(df: pd.DataFrame) -> pd.DataFrame:
"""Hourly volatility distribution — which hours move most."""
df = df.copy()
df["hour"] = df.index.hour
df["range"] = df["high"] - df["low"]
return df.groupby("hour").agg(
avg_range=("range", lambda x: round(x.mean() * 10000, 1)),
max_range=("range", lambda x: round(x.max() * 10000, 1)),
)