305 lines
12 KiB
Python
305 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ORB (Opening Range Breakout) Monitor for Futures
|
||
=================================================
|
||
|
||
Monitors session opens across global markets and detects ORB signals.
|
||
|
||
Sessions monitored (ET / UTC-4):
|
||
• Asia (Tokyo) — 19:00 ET (23:00 UTC)
|
||
• London — 03:00 ET (07:00 UTC)
|
||
• New York — 09:30 ET (13:30 UTC)
|
||
|
||
Strategy: After the session opens, the opening range high/low is captured
|
||
over `range_minutes`. If price later breaks above/below that range, an
|
||
ORB signal is logged.
|
||
|
||
⚠️ EDUCATIONAL PURPOSE ONLY — Not financial advice.
|
||
Uses yfinance (delayed data). NOT suitable for live trading.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
import logging
|
||
from datetime import datetime, timedelta, timezone
|
||
|
||
import yaml
|
||
import pandas as pd
|
||
import yfinance as yf
|
||
|
||
# ─── Constants ────────────────────────────────────────────────────────────────
|
||
|
||
# Yahoo Finance futures tickers
|
||
SYMBOLS = {
|
||
"ES": {"ticker": "ES=F", "name": "S&P 500 E-mini", "multiplier": 0.25},
|
||
"NQ": {"ticker": "NQ=F", "name": "Nasdaq 100 E-mini", "multiplier": 0.25},
|
||
"YM": {"ticker": "YM=F", "name": "Dow E-mini", "multiplier": 0.05},
|
||
"CL": {"ticker": "CL=F", "name": "Crude Oil WTI", "multiplier": 0.01},
|
||
"GC": {"ticker": "GC=F", "name": "Gold", "multiplier": 0.10},
|
||
}
|
||
|
||
# Session open times in UTC (no DST ambiguity)
|
||
SESSIONS = {
|
||
"asia": {"name": "Asia (Tokyo)", "open_utc": 23, "offset_min": 0},
|
||
"london": {"name": "London", "open_utc": 7, "offset_min": 0},
|
||
"ny": {"name": "New York", "open_utc": 13, "offset_min": 30},
|
||
}
|
||
|
||
UTC = timezone.utc
|
||
|
||
logger = logging.getLogger("ORB")
|
||
|
||
|
||
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
||
|
||
def load_config(path: str = "config.yaml") -> dict:
|
||
"""Load YAML configuration."""
|
||
try:
|
||
with open(path) as fh:
|
||
return yaml.safe_load(fh)
|
||
except FileNotFoundError:
|
||
logger.warning("config.yaml not found — using defaults")
|
||
return {}
|
||
|
||
|
||
|
||
|
||
|
||
def session_utc_start(date: datetime, session: dict) -> datetime:
|
||
"""Return UTC datetime when this session opens on the given UTC date."""
|
||
return datetime(date.year, date.month, date.day,
|
||
session["open_utc"], session["offset_min"],
|
||
tzinfo=UTC)
|
||
|
||
|
||
def fetch_data(ticker: str, days: int = 5) -> pd.DataFrame:
|
||
"""Fetch intraday futures data from Yahoo Finance (1-min bars)."""
|
||
end = datetime.now(UTC)
|
||
start = end - timedelta(days=days)
|
||
try:
|
||
df = yf.download(ticker, start=start, end=end,
|
||
interval="1m", progress=False, auto_adjust=True)
|
||
if df.empty:
|
||
logger.warning(f"No data returned for {ticker}")
|
||
return df
|
||
|
||
# Flatten MultiIndex columns (yf sometimes returns ('Close', ticker), etc.)
|
||
if isinstance(df.columns, pd.MultiIndex):
|
||
df.columns = [col[0] for col in df.columns]
|
||
|
||
return df
|
||
except Exception as e:
|
||
logger.error(f"Failed to fetch {ticker}: {e}")
|
||
return pd.DataFrame()
|
||
|
||
|
||
def find_session_bars(df: pd.DataFrame, session_start: datetime,
|
||
range_minutes: int) -> pd.DataFrame | None:
|
||
"""Extract the opening-range bars for a session, if data exists."""
|
||
# Allow ±2 min tolerance for session start
|
||
tolerance = timedelta(minutes=2)
|
||
end_bound = session_start + timedelta(minutes=range_minutes) + tolerance
|
||
mask = (df.index >= session_start - tolerance) & \
|
||
(df.index < end_bound)
|
||
range_bars = df.loc[mask]
|
||
return range_bars if len(range_bars) >= 5 else None # Need meaningful data
|
||
|
||
|
||
def analyze_orb(symbol_key: str, symbol_info: dict, df: pd.DataFrame,
|
||
session_key: str, session_info: dict,
|
||
cfg_orb: dict, date: datetime) -> list[dict]:
|
||
"""Check for ORB signals in the data for a given session date."""
|
||
range_min = cfg_orb.get("range_minutes", 30)
|
||
min_range = cfg_orb.get("min_range_ticks", 4)
|
||
max_range = cfg_orb.get("max_range_ticks", 100)
|
||
multiplier = symbol_info["multiplier"]
|
||
|
||
signals: list[dict] = []
|
||
session_start = session_utc_start(date, session_info)
|
||
|
||
# Try both start date and day before (in case of overnight sessions)
|
||
for offset in [0, -1]:
|
||
check_date = date + timedelta(days=offset)
|
||
try_start = datetime(check_date.year, check_date.month, check_date.day,
|
||
session_info["open_utc"], session_info["offset_min"],
|
||
tzinfo=UTC)
|
||
range_bars = find_session_bars(df, try_start, range_min)
|
||
if range_bars is None:
|
||
continue
|
||
|
||
# Opening range high/low — force scalar extraction
|
||
range_high = range_bars["High"].max().item()
|
||
range_low = range_bars["Low"].min().item()
|
||
range_size = range_high - range_low
|
||
range_ticks = range_size / multiplier
|
||
|
||
if range_ticks < min_range or range_ticks > max_range:
|
||
continue # Skip — range too small or too large
|
||
|
||
# Look for breakout in remaining data after range period
|
||
# Skip NaN rows and only use real data
|
||
remaining = df.loc[range_bars.index[-1]:].dropna(subset=["Close"])
|
||
if remaining.empty:
|
||
continue
|
||
|
||
# Bullish breakout: price closes above range high
|
||
bullish_bars = remaining[remaining["Close"] > range_high]
|
||
if not bullish_bars.empty:
|
||
breakout_time = bullish_bars.index[0]
|
||
breakout_price = float(bullish_bars.loc[breakout_time, "Close"])
|
||
signals.append({
|
||
"symbol": symbol_key,
|
||
"name": symbol_info["name"],
|
||
"session": session_info["name"],
|
||
"direction": "LONG",
|
||
"range_high": round(range_high, 2),
|
||
"range_low": round(range_low, 2),
|
||
"range_size": round(range_size, 2),
|
||
"range_ticks": round(range_ticks, 1),
|
||
"breakout_time": breakout_time,
|
||
"breakout_price": round(breakout_price, 2),
|
||
})
|
||
|
||
# Bearish breakout: price closes below range low
|
||
bearish_bars = remaining[remaining["Close"] < range_low]
|
||
if not bearish_bars.empty:
|
||
breakout_time = bearish_bars.index[0]
|
||
breakout_price = float(bearish_bars.loc[breakout_time, "Close"])
|
||
signals.append({
|
||
"symbol": symbol_key,
|
||
"name": symbol_info["name"],
|
||
"session": session_info["name"],
|
||
"direction": "SHORT",
|
||
"range_high": round(range_high, 2),
|
||
"range_low": round(range_low, 2),
|
||
"range_size": round(range_size, 2),
|
||
"range_ticks": round(range_ticks, 1),
|
||
"breakout_time": breakout_time,
|
||
"breakout_price": round(breakout_price, 2),
|
||
})
|
||
|
||
return signals
|
||
|
||
|
||
# ─── Main ─────────────────────────────────────────────────────────────────────
|
||
|
||
def run(date_str: str | None = None, days: int = 5,
|
||
config_path: str = "config.yaml") -> list[dict]:
|
||
"""
|
||
Analyze ORB patterns for all sessions and symbols.
|
||
|
||
Args:
|
||
date_str: Optional date in YYYY-MM-DD format. If None, uses today.
|
||
days: How many days of history to fetch.
|
||
config_path: Path to config.yaml.
|
||
|
||
Returns:
|
||
List of signal dicts sorted by breakout_time.
|
||
"""
|
||
cfg = load_config(config_path)
|
||
cfg_orb = cfg.get("orb", {})
|
||
|
||
target_date = datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=UTC) if date_str else datetime.now(UTC)
|
||
# Expand search window to cover all sessions around the target date
|
||
search_dates = [target_date + timedelta(days=d) for d in range(-1, days)]
|
||
|
||
all_signals: list[dict] = []
|
||
symbol_items = cfg.get("symbols", SYMBOLS) or SYMBOLS
|
||
|
||
for sym_key, sym_info in symbol_items.items():
|
||
ticker = sym_info.get("ticker", f"{sym_key}=F")
|
||
print(f" Fetching {ticker} ({sym_info.get('name', sym_key)}) …", flush=True)
|
||
df = fetch_data(ticker, days=days)
|
||
if df.empty:
|
||
continue
|
||
|
||
# Reconcile multiplier from config vs hardcoded
|
||
sym_info.setdefault("multiplier", SYMBOLS.get(sym_key, {}).get("multiplier", 0.25))
|
||
|
||
for sd in search_dates:
|
||
for sess_key, sess_info in SESSIONS.items():
|
||
signals = analyze_orb(
|
||
sym_key, sym_info, df,
|
||
sess_key, sess_info, cfg_orb, sd
|
||
)
|
||
all_signals.extend(signals)
|
||
|
||
# Sort by breakout time
|
||
all_signals.sort(key=lambda s: s["breakout_time"])
|
||
return all_signals
|
||
|
||
|
||
def format_report(signals: list[dict]) -> str:
|
||
"""Pretty-print ORB signals for Telegram / terminal."""
|
||
if not signals:
|
||
return (
|
||
"📊 **ORB Scan Complete — No Signals Found**\n\n"
|
||
"No opening range breakouts detected in the scanned period.\n"
|
||
"The market may be quiet, or the data may be too delayed.\n\n"
|
||
"*Run again closer to session opens for best results.*"
|
||
)
|
||
|
||
lines = [
|
||
f"📊 **ORB Signals Found** ({len(signals)} signals)",
|
||
f"_Scanned: {datetime.now(UTC).strftime('%Y-%m-%d %H:%M UTC')}_",
|
||
"─" * 40,
|
||
]
|
||
|
||
for s in signals:
|
||
direction = "🟢 LONG" if s["direction"] == "LONG" else "🔴 SHORT"
|
||
lines.append(
|
||
f"\n**{s['symbol']}** ({s['name']}) — {s['session']}\n"
|
||
f"{direction}\n"
|
||
f" Range: {s['range_low']} – {s['range_high']} "
|
||
f"({s['range_size']} pts / {s['range_ticks']} ticks)\n"
|
||
f" Breakout: {s['breakout_price']} at "
|
||
f"`{s['breakout_time'].strftime('%H:%M UTC')}`"
|
||
)
|
||
|
||
lines.append("\n" + "─" * 40)
|
||
lines.append(
|
||
"⚠️ _Educational analysis only. Uses delayed data._\n"
|
||
"_Not financial advice. Verify with live data before trading._"
|
||
)
|
||
return "\n".join(lines)
|
||
|
||
|
||
# ─── CLI ──────────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
|
||
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="ORB Futures Monitor")
|
||
parser.add_argument("--date", type=str, default=None,
|
||
help="Target date YYYY-MM-DD (default: today)")
|
||
parser.add_argument("--days", type=int, default=5,
|
||
help="Days of history to scan (default: 5)")
|
||
parser.add_argument("--config", type=str, default="config.yaml",
|
||
help="Path to config file")
|
||
parser.add_argument("--json", action="store_true",
|
||
help="Output as JSON instead of formatted text")
|
||
args = parser.parse_args()
|
||
|
||
print("\n🔍 ORB Monitor — Scanning futures data …\n", flush=True)
|
||
signals = run(date_str=args.date, days=args.days, config_path=args.config)
|
||
|
||
if args.json:
|
||
import json
|
||
print(json.dumps(signals, indent=2, default=str))
|
||
else:
|
||
report = format_report(signals)
|
||
print(report)
|
||
# Also save to log
|
||
log_path = "orb_signals.log"
|
||
with open(log_path, "a") as fh:
|
||
fh.write(f"\n{'='*50}\n")
|
||
fh.write(f"Scan: {datetime.now(UTC).isoformat()}\n")
|
||
fh.write(report + "\n")
|
||
|
||
print(f"\n✅ Done. {len(signals)} signals detected.\n", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|