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)),
        )

Usage: Check session context before every trade. A breakout strategy at 3AM UTC in EURUSD will fail — save it for London open.