diff --git a/data_provider.py b/data_provider.py index 0fdede4..ab0b4ab 100644 --- a/data_provider.py +++ b/data_provider.py @@ -1,19 +1,21 @@ """ Forex Signal Intelligence Platform - Data Provider ==================================================== -Retrieves and validates market data from multiple sources. -Uses Yahoo Finance as the free data backbone with integrity checks. +Retrieves and validates market data from Twelve Data API. +Falls back to synthetic GBM data if the API is unavailable. """ +import hashlib import logging +import os import time -from datetime import datetime, timedelta -from typing import Dict, List, Optional, Tuple +from collections import deque +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional import numpy as np import pandas as pd import requests -import yfinance as yf from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY @@ -27,46 +29,61 @@ class DataIntegrityError(Exception): class ForexDataProvider: """ - Multi-source Forex data provider with validation. - - Uses Yahoo Finance (free, no API key) as primary source. - Implements data integrity checks to catch bad/stale data before - it reaches the signal engine. + Forex data provider backed by Twelve Data API. + + Twelve Data has native 4h candles, eliminating the H1→H4 resampling + hack needed for Yahoo Finance. Free tier: 800 credits/day, 8 req/min. """ - # Yahoo Finance timeframe mapping - YF_INTERVALS = { - Timeframe.M1: "1m", - Timeframe.M5: "5m", - Timeframe.M15: "15m", - Timeframe.M30: "30m", - Timeframe.H1: "60m", - Timeframe.H4: "60m", # aggregate from H1 - Timeframe.D1: "1d", - Timeframe.W1: "1wk", + TD_BASE_URL = "https://api.twelvedata.com" + + TD_INTERVALS = { + Timeframe.M1: "1min", + Timeframe.M5: "5min", + Timeframe.M15: "15min", + Timeframe.M30: "30min", + Timeframe.H1: "1h", + Timeframe.H4: "4h", + Timeframe.D1: "1day", + Timeframe.W1: "1week", } - # Max lookback per interval (Yahoo limits) - YF_MAX_RANGE = { - Timeframe.M1: 7, - Timeframe.M5: 60, - Timeframe.M15: 60, - Timeframe.M30: 60, - Timeframe.H1: 730, - Timeframe.H4: 730, - Timeframe.D1: 3650, - Timeframe.W1: 3650, + # Approximate bars per day for each timeframe (used to size outputsize) + BARS_PER_DAY = { + Timeframe.M1: 1440, + Timeframe.M5: 288, + Timeframe.M15: 96, + Timeframe.M30: 48, + Timeframe.H1: 24, + Timeframe.H4: 6, + Timeframe.D1: 1, + Timeframe.W1: 0.143, } + # Free plan: 8 API credits per minute (each symbol = 1 credit) + RATE_LIMIT = 8 + RATE_WINDOW = 60.0 + def __init__(self, config: PlatformConfig): self.config = config - self._cache: Dict[str, Dict[str, pd.DataFrame]] = {} + self._api_key = ( + os.environ.get("TWELVE_DATA_API_KEY") + or config.twelve_data_api_key + ) + self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} + self._request_times: deque = deque() # timestamps of recent API credits used self._session = requests.Session() self._session.headers.update({ - "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36" + "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)" }) + @staticmethod + def _to_td_symbol(symbol: str) -> str: + """Convert internal symbol (EURUSD=X) to Twelve Data format (EUR/USD).""" + clean = symbol.replace("=X", "") + return f"{clean[:3]}/{clean[3:]}" + def fetch_ohlcv( self, symbol: str, @@ -75,22 +92,16 @@ class ForexDataProvider: ) -> pd.DataFrame: """ Fetch OHLCV data for a Forex pair with full validation. - + Returns a DataFrame with columns: Open, High, Low, Close, Volume - Index is DatetimeIndex in UTC. - - Raises DataIntegrityError if data is invalid/insufficient. + Index is a timezone-aware DatetimeIndex in UTC. """ if lookback_days is None: - lookback_days = min( - self.config.data_lookback_days, - self.YF_MAX_RANGE[timeframe], - ) + lookback_days = self.config.data_lookback_days cache_key = f"{symbol}_{timeframe.value}" now = time.time() - # Check cache (respect refresh interval) if ( cache_key in self._cache_timestamps and (now - self._cache_timestamps[cache_key]) < self.config.refresh_interval @@ -100,21 +111,37 @@ class ForexDataProvider: if cached is not None and not cached.empty: return cached - df = self._fetch_yahoo(symbol, timeframe, lookback_days) - - # If H4, resample from H1 data - if timeframe == Timeframe.H4: - df = self._resample_to_h4(df) - - # Validate data integrity + df = self._fetch_twelve_data(symbol, timeframe, lookback_days) df = self._validate_and_clean(df, symbol, timeframe) - # Cache it self._cache[cache_key] = df self._cache_timestamps[cache_key] = now - return df + def prefetch_batch( + self, + symbols: List[str], + timeframes: List[Timeframe], + lookback_days: Optional[int] = None, + ) -> None: + """ + Batch-fetch all symbols for each timeframe in a single API call per TF. + + Twelve Data accepts comma-separated symbols, so 7 pairs × 3 timeframes + = 3 requests instead of 21 — well within the free-tier 8 req/min limit. + Results are stored in the local cache; subsequent fetch_ohlcv calls + for the same symbol/TF will be served from cache. + """ + if lookback_days is None: + lookback_days = self.config.data_lookback_days + + for tf in timeframes: + try: + self._batch_fetch_timeframe(symbols, tf, lookback_days) + time.sleep(1.0) # brief pause between TF requests + except Exception as e: + logger.warning(f"Batch prefetch failed for {tf.value}: {e}") + def fetch_multiple( self, symbols: List[str], @@ -127,14 +154,12 @@ class ForexDataProvider: try: df = self.fetch_ohlcv(symbol, timeframe, lookback_days) results[symbol] = df - time.sleep(0.3) # rate limit courtesy except Exception as e: logger.warning(f"Failed to fetch {symbol}: {e}") - continue return results def get_latest_price(self, symbol: str) -> Optional[float]: - """Get the most recent close price for a symbol.""" + """Return the most recent close price for a symbol.""" try: df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) if df is not None and not df.empty: @@ -144,20 +169,14 @@ class ForexDataProvider: return None def get_spread_estimate(self, symbol: str) -> float: - """ - Estimate the typical spread for a pair based on recent data. - Uses high-low range of 1-minute bars as a proxy. - """ + """Estimate typical spread from recent M5 high-low ranges.""" try: df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) if df is not None and not df.empty: - # Median range as spread estimate (conservative) - ranges = df["High"] - df["Low"] - return float(ranges.median()) + return float((df["High"] - df["Low"]).median()) except Exception: pass - # Fallback: typical spread estimates for majors fallback_spreads = { "EURUSD=X": 0.00010, "GBPUSD=X": 0.00015, "USDJPY=X": 0.010, "USDCHF=X": 0.00015, @@ -166,88 +185,220 @@ class ForexDataProvider: } return fallback_spreads.get(symbol, 0.00020) + def get_pair_info(self, symbol: str) -> Dict: + """Return display name and pip value for a pair.""" + display = SYMBOL_DISPLAY.get(symbol, symbol) + is_jpy = "JPY" in symbol + return { + "symbol": symbol, + "display": display, + "pip_value": 0.01 if is_jpy else 0.0001, + "is_jpy_pair": is_jpy, + } + # ── Private Methods ──────────────────────────────────────────── - def _fetch_yahoo( + def _throttle(self, credits: int = 1) -> None: + """ + Block until `credits` API credits are available. + + Twelve Data free tier: 8 credits/minute. Each symbol in any request + (single or batch) consumes 1 credit. This method enforces the limit + regardless of whether requests are batched or individual. + """ + while True: + now = time.time() + # Evict credits older than the rate window + while self._request_times and now - self._request_times[0] >= self.RATE_WINDOW: + self._request_times.popleft() + + available = self.RATE_LIMIT - len(self._request_times) + if available >= credits: + for _ in range(credits): + self._request_times.append(time.time()) + return + + # Wait until the oldest credit expires + wait = (self._request_times[0] + self.RATE_WINDOW) - now + 0.2 + logger.info( + f"Rate limit: {len(self._request_times)}/{self.RATE_LIMIT} credits used " + f"— waiting {wait:.1f}s for next window..." + ) + time.sleep(wait) + + def _batch_fetch_timeframe( + self, symbols: List[str], timeframe: Timeframe, lookback_days: int + ) -> None: + """ + Single API call fetching all symbols for one timeframe. + Twelve Data returns a dict keyed by symbol when multiple are requested. + Parsed DataFrames are validated and stored in the cache. + """ + if not self._api_key: + return + + td_symbols = [self._to_td_symbol(s) for s in symbols] + interval = self.TD_INTERVALS[timeframe] + outputsize = min( + int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000 + ) + outputsize = max(outputsize, 300) + + # Each symbol in a batch request costs 1 credit + self._throttle(credits=len(symbols)) + + resp = self._session.get( + f"{self.TD_BASE_URL}/time_series", + params={ + "symbol": ",".join(td_symbols), + "interval": interval, + "outputsize": outputsize, + "apikey": self._api_key, + "timezone": "UTC", + "format": "json", + }, + timeout=30, + ) + resp.raise_for_status() + data = resp.json() + + # If the whole batch request returned a top-level error, raise it + if isinstance(data, dict) and data.get("status") == "error": + raise DataIntegrityError( + f"Batch API error ({timeframe.value}): {data.get('message', 'unknown')}" + ) + + # Batch response: {"EUR/USD": {...}, "GBP/USD": {...}, ...} + now = time.time() + for symbol, td_sym in zip(symbols, td_symbols): + try: + sym_data = data.get(td_sym, {}) + if sym_data.get("status") == "error": + logger.warning( + f"Batch fetch error for {td_sym}: {sym_data.get('message')}" + ) + continue + + values = sym_data.get("values") + if not values: + logger.warning(f"No values in batch response for {td_sym}") + continue + + records = [ + { + "Datetime": pd.Timestamp(v["datetime"], tz="UTC"), + "Open": float(v["open"]), + "High": float(v["high"]), + "Low": float(v["low"]), + "Close": float(v["close"]), + "Volume": float(v.get("volume") or 0), + } + for v in reversed(values) + ] + df = pd.DataFrame(records).set_index("Datetime") + df.index.name = "Datetime" + df = self._validate_and_clean(df, symbol, timeframe) + + cache_key = f"{symbol}_{timeframe.value}" + self._cache[cache_key] = df + self._cache_timestamps[cache_key] = now + logger.debug(f"Batch cached {len(df)} bars for {td_sym} @ {interval}") + + except Exception as e: + logger.warning(f"Failed to parse batch data for {td_sym}: {e}") + + def _fetch_twelve_data( self, symbol: str, timeframe: Timeframe, lookback_days: int ) -> pd.DataFrame: - """Fetch raw OHLCV from Yahoo Finance via yfinance library. - Falls back to realistic synthetic data if network unavailable.""" - interval = self.YF_INTERVALS[timeframe] - period_map = { - 1: "1d", 2: "2d", 5: "5d", 7: "5d", - 30: "1mo", 60: "3mo", 90: "3mo", - 180: "6mo", 365: "1y", 730: "2y", - } - best_period = "3mo" - for days, period in sorted(period_map.items()): - if lookback_days <= days: - best_period = period - break + """ + Fetch OHLCV from Twelve Data REST API. + Falls back to synthetic GBM data if API is unavailable or key is missing. + """ + if not self._api_key: + logger.warning("No Twelve Data API key — using synthetic data. " + "Set TWELVE_DATA_API_KEY env var or config.twelve_data_api_key.") + return self._generate_synthetic_data(symbol, timeframe, lookback_days) + + td_symbol = self._to_td_symbol(symbol) + interval = self.TD_INTERVALS[timeframe] + outputsize = min( + int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, + 5000, + ) + outputsize = max(outputsize, 300) + + self._throttle(credits=1) try: - ticker = yf.Ticker(symbol) - df = ticker.history(period=best_period, interval=interval) + resp = self._session.get( + f"{self.TD_BASE_URL}/time_series", + params={ + "symbol": td_symbol, + "interval": interval, + "outputsize": outputsize, + "apikey": self._api_key, + "timezone": "UTC", + "format": "json", + }, + timeout=15, + ) + resp.raise_for_status() + data = resp.json() - if df is not None and not df.empty: - # Standardize column names - col_map = {} - for col in df.columns: - lower = col.lower() - if lower == "open": col_map[col] = "Open" - elif lower == "high": col_map[col] = "High" - elif lower == "low": col_map[col] = "Low" - elif lower == "close": col_map[col] = "Close" - elif lower == "volume": col_map[col] = "Volume" - df = df.rename(columns=col_map) + if data.get("status") == "error": + raise DataIntegrityError( + f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}" + ) - for req in ["Open", "High", "Low", "Close"]: - if req not in df.columns: - raise DataIntegrityError(f"Missing {req} column for {symbol}") + values = data.get("values") + if not values: + raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") - if "Volume" not in df.columns: - df["Volume"] = 0 + # Values arrive newest-first; reverse to chronological order + records = [ + { + "Datetime": pd.Timestamp(v["datetime"], tz="UTC"), + "Open": float(v["open"]), + "High": float(v["high"]), + "Low": float(v["low"]), + "Close": float(v["close"]), + "Volume": float(v.get("volume") or 0), + } + for v in reversed(values) + ] - if df.index.tz is None: - df.index = df.index.tz_localize("UTC") - else: - df.index = df.index.tz_convert("UTC") + df = pd.DataFrame(records).set_index("Datetime") + df.index.name = "Datetime" + logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") + return df - df.index.name = "Datetime" - df = df[["Open", "High", "Low", "Close", "Volume"]] - return df + except DataIntegrityError: + raise except Exception as e: - logger.info(f"Live data unavailable for {symbol}, using synthetic: {e}") - - # ── Fallback: Realistic synthetic Forex data ─────────────── - return self._generate_synthetic_data(symbol, timeframe, lookback_days) + logger.info(f"Twelve Data fetch failed for {symbol} ({e}), using synthetic data.") + return self._generate_synthetic_data(symbol, timeframe, lookback_days) def _generate_synthetic_data( self, symbol: str, timeframe: Timeframe, lookback_days: int ) -> pd.DataFrame: """ - Generate realistic synthetic Forex OHLCV data. - - Uses GBM (Geometric Brownian Motion) with mean-reversion, - realistic volatility per pair, and proper OHLC construction. - This ensures the technical engine produces meaningful signals - even without live data. + Generate realistic synthetic Forex OHLCV using GBM with mean-reversion. + Used as a fallback when live data is unavailable. """ - # Realistic base prices and volatilities (annualized) for major pairs pair_params = { "EURUSD=X": {"price": 1.1350, "vol": 0.07, "drift": -0.001}, - "GBPUSD=X": {"price": 1.3200, "vol": 0.08, "drift": 0.002}, - "USDJPY=X": {"price": 144.50, "vol": 0.09, "drift": 0.003}, + "GBPUSD=X": {"price": 1.3200, "vol": 0.08, "drift": 0.002}, + "USDJPY=X": {"price": 144.50, "vol": 0.09, "drift": 0.003}, "USDCHF=X": {"price": 0.8850, "vol": 0.07, "drift": -0.001}, "AUDUSD=X": {"price": 0.6550, "vol": 0.10, "drift": -0.002}, - "USDCAD=X": {"price": 1.3650, "vol": 0.07, "drift": 0.001}, + "USDCAD=X": {"price": 1.3650, "vol": 0.07, "drift": 0.001}, "NZDUSD=X": {"price": 0.6100, "vol": 0.10, "drift": -0.001}, - "EURGBP=X": {"price": 0.8590, "vol": 0.06, "drift": 0.000}, - "EURJPY=X": {"price": 163.90, "vol": 0.09, "drift": 0.002}, - "GBPJPY=X": {"price": 190.70, "vol": 0.11, "drift": 0.003}, - "AUDJPY=X": {"price": 94.60, "vol": 0.11, "drift": 0.001}, - "EURCHF=X": {"price": 0.9380, "vol": 0.05, "drift": 0.000}, - "CADJPY=X": {"price": 105.80, "vol": 0.09, "drift": 0.001}, + "EURGBP=X": {"price": 0.8590, "vol": 0.06, "drift": 0.000}, + "EURJPY=X": {"price": 163.90, "vol": 0.09, "drift": 0.002}, + "GBPJPY=X": {"price": 190.70, "vol": 0.11, "drift": 0.003}, + "AUDJPY=X": {"price": 94.60, "vol": 0.11, "drift": 0.001}, + "EURCHF=X": {"price": 0.9380, "vol": 0.05, "drift": 0.000}, + "CADJPY=X": {"price": 105.80, "vol": 0.09, "drift": 0.001}, } params = pair_params.get(symbol, {"price": 1.0000, "vol": 0.08, "drift": 0.000}) @@ -255,7 +406,6 @@ class ForexDataProvider: annual_vol = params["vol"] annual_drift = params["drift"] - # Calculate bar parameters tf_minutes = { Timeframe.M1: 1, Timeframe.M5: 5, Timeframe.M15: 15, Timeframe.M30: 30, Timeframe.H1: 60, Timeframe.H4: 240, @@ -266,40 +416,29 @@ class ForexDataProvider: dt = 1.0 / bars_per_year n_bars = min(int(lookback_days * 24 * 60 / bar_minutes), 2000) - n_bars = max(n_bars, 300) # minimum for indicators + n_bars = max(n_bars, 300) - # Use a seed based on symbol for reproducibility within a session - # but add time-based component so it changes between runs - import hashlib seed_base = int(hashlib.md5(symbol.encode()).hexdigest()[:8], 16) - time_component = int(time.time()) // 3600 # changes hourly + time_component = int(time.time()) // 3600 np.random.seed((seed_base + time_component) % (2**31)) - # Generate returns using GBM with mean reversion bar_vol = annual_vol * np.sqrt(dt) bar_drift = annual_drift * dt - # Add trending behavior (occasional trends + mean reversion) regime_len = np.random.randint(20, 80) trend_strength = np.random.normal(0, 0.3) - - returns = np.zeros(n_bars) + price = base_price prices = [price] for i in range(1, n_bars): - # Regime changes if i % regime_len == 0: regime_len = np.random.randint(20, 80) trend_strength = np.random.normal(0, 0.3) - # Mean reversion component mr_force = -0.001 * (price - base_price) / base_price - - # Combined return ret = bar_drift + trend_strength * dt + mr_force + bar_vol * np.random.randn() - - # Occasional volatility spikes (simulates news events) + if np.random.random() < 0.02: ret *= np.random.uniform(2.0, 4.0) @@ -307,101 +446,63 @@ class ForexDataProvider: prices.append(price) prices = np.array(prices) - - # Construct OHLC from close prices opens = np.roll(prices, 1) opens[0] = prices[0] - # Realistic High/Low generation typical_range = bar_vol * base_price * 1.5 highs = np.maximum(prices, opens) + np.abs(np.random.normal(0, typical_range * 0.6, n_bars)) - lows = np.minimum(prices, opens) - np.abs(np.random.normal(0, typical_range * 0.6, n_bars)) - - # Ensure OHLC consistency + lows = np.minimum(prices, opens) - np.abs(np.random.normal(0, typical_range * 0.6, n_bars)) highs = np.maximum(highs, np.maximum(opens, prices)) - lows = np.minimum(lows, np.minimum(opens, prices)) + lows = np.minimum(lows, np.minimum(opens, prices)) - # Volume (tick volume approximation) - base_vol = np.random.lognormal(mean=10, sigma=0.5, size=n_bars) - # Higher volume during London/NY overlap hours - volume = base_vol.astype(int) + volume = np.random.lognormal(mean=10, sigma=0.5, size=n_bars).astype(int) - # Generate timestamps - end_time = pd.Timestamp.now(tz="UTC").floor(f"{bar_minutes}min") freq_map = { - Timeframe.M1: "1min", Timeframe.M5: "5min", + Timeframe.M1: "1min", Timeframe.M5: "5min", Timeframe.M15: "15min", Timeframe.M30: "30min", - Timeframe.H1: "1h", Timeframe.H4: "4h", - Timeframe.D1: "1D", Timeframe.W1: "1W", + Timeframe.H1: "1h", Timeframe.H4: "4h", + Timeframe.D1: "D", Timeframe.W1: "W", } freq = freq_map.get(timeframe, "1h") - - # Create business-hours aware index - index = pd.date_range( - end=end_time, periods=n_bars, freq=freq, tz="UTC" - ) + end_time = pd.Timestamp.now(tz="UTC").floor(f"{bar_minutes}min") + index = pd.date_range(end=end_time, periods=n_bars, freq=freq, tz="UTC") df = pd.DataFrame({ - "Open": opens, - "High": highs, - "Low": lows, - "Close": prices, - "Volume": volume, + "Open": opens, "High": highs, "Low": lows, + "Close": prices, "Volume": volume, }, index=index) - df.index.name = "Datetime" logger.info( f"Generated {n_bars} synthetic {timeframe.value} bars for " f"{SYMBOL_DISPLAY.get(symbol, symbol)} @ {prices[-1]:.5f}" ) - return df - def _resample_to_h4(self, df: pd.DataFrame) -> pd.DataFrame: - """Resample H1 data to H4 candles.""" - resampled = df.resample("4h").agg({ - "Open": "first", - "High": "max", - "Low": "min", - "Close": "last", - "Volume": "sum", - }).dropna() - return resampled - def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe ) -> pd.DataFrame: """ - Comprehensive data validation and cleaning. - - Checks: - 1. Sufficient data points for indicator calculation - 2. No future timestamps - 3. OHLC logical consistency (High >= Low, etc.) - 4. No extreme outliers (>10 sigma moves in a single bar) - 5. Forward-fill small gaps, reject large ones + Validate and clean OHLCV data. + + Steps: drop all-NaN rows → ffill small gaps → remove future timestamps + → fix OHLC inconsistencies → remove >10-sigma outliers → sort & deduplicate. """ if df is None or df.empty: raise DataIntegrityError(f"Empty dataset for {symbol}") original_len = len(df) - # 1. Drop rows with all NaN OHLC df = df.dropna(subset=["Open", "High", "Low", "Close"], how="all") - - # 2. Forward-fill small gaps (up to 3 consecutive NaN bars) df = df.ffill(limit=3) df = df.dropna(subset=["Close"]) - # 3. No future timestamps now = pd.Timestamp.utcnow() future_mask = df.index > now + pd.Timedelta(minutes=5) if future_mask.any(): logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps") df = df[~future_mask] - # 4. OHLC logical consistency invalid_ohlc = ( (df["High"] < df["Low"]) | (df["High"] < df["Open"]) | @@ -410,31 +511,25 @@ class ForexDataProvider: (df["Low"] > df["Close"]) ) if invalid_ohlc.any(): - count = invalid_ohlc.sum() - logger.warning(f"{symbol}: Fixing {count} OHLC-inconsistent bars") - # Fix by recalculating High/Low from OHLC + logger.warning(f"{symbol}: Fixing {invalid_ohlc.sum()} OHLC-inconsistent bars") for idx in df[invalid_ohlc].index: vals = [df.loc[idx, "Open"], df.loc[idx, "Close"]] df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]]) - df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) + df.loc[idx, "Low"] = min(vals + [df.loc[idx, "Low"]]) - # 5. Outlier detection (>10 sigma single-bar return) returns = df["Close"].pct_change() - mean_ret = returns.mean() std_ret = returns.std() if std_ret > 0: - outlier_mask = (returns - mean_ret).abs() > 10 * std_ret + outlier_mask = (returns - returns.mean()).abs() > 10 * std_ret if outlier_mask.any(): - count = outlier_mask.sum() - logger.warning(f"{symbol}: Removing {count} extreme outlier bars") + logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars") df = df[~outlier_mask] - # 6. Minimum data check min_required = max( self.config.technical.ema_trend, self.config.technical.ichimoku_senkou_b, 200, - ) + 10 # buffer + ) + 10 if len(df) < min_required: logger.warning( @@ -442,10 +537,7 @@ class ForexDataProvider: f"(need {min_required}). Some indicators will be partial." ) - # 7. Sort by time (paranoia check) df = df.sort_index() - - # 8. Remove duplicate timestamps df = df[~df.index.duplicated(keep="last")] cleaned = original_len - len(df) @@ -453,16 +545,3 @@ class ForexDataProvider: logger.info(f"{symbol}: Cleaned {cleaned}/{original_len} bars") return df - - def get_pair_info(self, symbol: str) -> Dict: - """Get display name and pip value for a pair.""" - display = SYMBOL_DISPLAY.get(symbol, symbol) - # JPY pairs have 2 decimal places, others have 4-5 - is_jpy = "JPY" in symbol - pip_value = 0.01 if is_jpy else 0.0001 - return { - "symbol": symbol, - "display": display, - "pip_value": pip_value, - "is_jpy_pair": is_jpy, - } \ No newline at end of file diff --git a/engine.py b/engine.py index 5b22069..046374e 100644 --- a/engine.py +++ b/engine.py @@ -13,7 +13,6 @@ Design philosophy: import logging from dataclasses import dataclass, field -from enum import Enum from typing import Dict, List, Optional, Tuple import numpy as np @@ -54,7 +53,7 @@ class TechnicalAnalysis: class TechnicalEngine: """ Core technical analysis engine. - + Computes all indicators on a given OHLCV DataFrame and produces a composite signal with confidence scoring. """ @@ -65,7 +64,7 @@ class TechnicalEngine: def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis: """ Run full technical analysis on OHLCV data. - + Returns TechnicalAnalysis with all indicators, composite signal, and market regime classification. """ @@ -123,7 +122,6 @@ class TechnicalEngine: f, m, s = ema_fast.iloc[-1], ema_medium.iloc[-1], ema_slow.iloc[-1] t = ema_trend.iloc[-1] if ema_trend is not None and len(ema_trend) >= c.ema_trend else None - # Perfect alignment scoring # Bullish: price > fast > medium > slow > trend # Bearish: price < fast < medium < slow < trend alignment_score = 0 @@ -190,7 +188,7 @@ class TechnicalEngine: rsi = 100 - (100 / (1 + rs)) rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0 - if np.isnan(rsi_val): + if pd.isna(rsi_val): rsi_val = 50.0 # Signal interpretation (context-aware) @@ -220,12 +218,8 @@ class TechnicalEngine: confidence=conf, description=desc, weight=1.0, ) - # Store RSI series for divergence detection - result.indicators["_rsi_series"] = IndicatorSignal( - name="_rsi_series", value=0, signal=SignalStrength.NEUTRAL, - confidence=0, description="", weight=0, - ) - result._rsi_series = rsi # attach for divergence use + # Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix) + result._rsi_series = rsi # type: ignore[attr-defined] def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis): """MACD with histogram momentum analysis.""" @@ -293,8 +287,8 @@ class TechnicalEngine: confidence=conf, description=desc, weight=1.1, ) - # Store for divergence - result._macd_histogram = histogram + # Attach histogram for divergence detection + result._macd_histogram = histogram # type: ignore[attr-defined] def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis): """Bollinger Bands position and squeeze detection.""" @@ -310,17 +304,24 @@ class TechnicalEngine: upper_val = upper.iloc[-1] lower_val = lower.iloc[-1] sma_val = sma.iloc[-1] - band_width = (upper_val - lower_val) / sma_val if sma_val != 0 else 0 + + # Guard against NaN (insufficient data) + if pd.isna(upper_val) or pd.isna(lower_val) or pd.isna(sma_val) or sma_val == 0: + return + + band_width = (upper_val - lower_val) / sma_val # Position within bands (0 = lower band, 1 = upper band) - bb_position = (latest - lower_val) / (upper_val - lower_val) if (upper_val - lower_val) != 0 else 0.5 + bb_range = upper_val - lower_val + bb_position = (latest - lower_val) / bb_range if bb_range != 0 else 0.5 # Squeeze detection (band width compared to recent average) + is_squeeze = False if len(std) >= 20: - avg_width = ((upper - lower) / sma).rolling(20).mean().iloc[-1] - is_squeeze = band_width < avg_width * 0.75 - else: - is_squeeze = False + avg_width_series = ((upper - lower) / sma).rolling(20).mean() + avg_width = avg_width_series.iloc[-1] + if not pd.isna(avg_width) and avg_width > 0: + is_squeeze = band_width < avg_width * 0.75 # Signal squeeze_note = " | SQUEEZE detected (expect breakout)" if is_squeeze else "" @@ -376,9 +377,9 @@ class TechnicalEngine: k_val = k_line.iloc[-1] if not k_line.empty else 50.0 d_val = d_line.iloc[-1] if not d_line.empty else 50.0 - if np.isnan(k_val): + if pd.isna(k_val): k_val = 50.0 - if np.isnan(d_val): + if pd.isna(d_val): d_val = 50.0 # Crossover detection @@ -386,7 +387,7 @@ class TechnicalEngine: if len(k_line) >= 2 and len(d_line) >= 2: prev_k = k_line.iloc[-2] prev_d = d_line.iloc[-2] - if not np.isnan(prev_k) and not np.isnan(prev_d): + if not pd.isna(prev_k) and not pd.isna(prev_d): if prev_k <= prev_d and k_val > d_val: crossover = " | Bullish %K/%D cross" elif prev_k >= prev_d and k_val < d_val: @@ -445,9 +446,13 @@ class TechnicalEngine: dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() - adx_val = adx.iloc[-1] if not adx.empty and not np.isnan(adx.iloc[-1]) else 0.0 - plus_di_val = plus_di.iloc[-1] if not plus_di.empty else 0.0 - minus_di_val = minus_di.iloc[-1] if not minus_di.empty else 0.0 + adx_val = adx.iloc[-1] if not adx.empty and not pd.isna(adx.iloc[-1]) else 0.0 + plus_di_last = plus_di.iloc[-1] if not plus_di.empty else np.nan + minus_di_last = minus_di.iloc[-1] if not minus_di.empty else np.nan + + # Guard NaN — treat as equal (no directional edge) if data is missing + plus_di_val = float(plus_di_last) if not pd.isna(plus_di_last) else 0.0 + minus_di_val = float(minus_di_last) if not pd.isna(minus_di_last) else 0.0 # Signal (ADX tells strength, DI tells direction) if adx_val >= c.adx_very_strong: @@ -486,6 +491,8 @@ class TechnicalEngine: atr = tr.ewm(alpha=1/c.atr_period, min_periods=c.atr_period).mean() atr_val = atr.iloc[-1] if not atr.empty else 0.0 + if pd.isna(atr_val): + atr_val = 0.0 # ATR as percentage of price atr_pct = (atr_val / close.iloc[-1]) * 100 if close.iloc[-1] != 0 else 0 @@ -513,7 +520,7 @@ class TechnicalEngine: name="ATR", value=atr_val, signal=SignalStrength.NEUTRAL, confidence=50, description=f"ATR {atr_val:.5f} ({atr_pct:.3f}%) | Volatility: {vol_regime} (P{atr_percentile:.0f})", - weight=0.0, # ATR doesn't produce buy/sell, it's context + weight=0.0, # ATR is context, not a directional signal ) def _compute_ichimoku(self, df: pd.DataFrame, result: TechnicalAnalysis): @@ -532,32 +539,26 @@ class TechnicalEngine: kijun = (high.rolling(c.ichimoku_kijun).max() + low.rolling(c.ichimoku_kijun).min()) / 2 - # Senkou Span A (Leading Span A) - senkou_a = ((tenkan + kijun) / 2).shift(c.ichimoku_kijun) - - # Senkou Span B (Leading Span B) - senkou_b = ((high.rolling(c.ichimoku_senkou_b).max() + - low.rolling(c.ichimoku_senkou_b).min()) / 2).shift(c.ichimoku_kijun) - latest = close.iloc[-1] tenkan_val = tenkan.iloc[-1] kijun_val = kijun.iloc[-1] - # Use non-shifted values for current cloud - sa = ((tenkan.iloc[-1] + kijun.iloc[-1]) / 2) + # Current (unshifted) cloud + sa = (tenkan.iloc[-1] + kijun.iloc[-1]) / 2 sb_raw = (high.rolling(c.ichimoku_senkou_b).max() + low.rolling(c.ichimoku_senkou_b).min()) / 2 sb = sb_raw.iloc[-1] + if pd.isna(sa) or pd.isna(sb) or pd.isna(tenkan_val) or pd.isna(kijun_val): + return + cloud_top = max(sa, sb) cloud_bottom = min(sa, sb) cloud_is_green = sa > sb - # Signal analysis signals = [] score = 0 - # Price vs Cloud if latest > cloud_top: signals.append("Price ABOVE cloud") score += 2 @@ -567,7 +568,6 @@ class TechnicalEngine: else: signals.append("Price IN cloud (indecision)") - # Tenkan vs Kijun if tenkan_val > kijun_val: signals.append("TK cross bullish") score += 1 @@ -575,7 +575,6 @@ class TechnicalEngine: signals.append("TK cross bearish") score -= 1 - # Cloud color if cloud_is_green: signals.append("Green cloud") score += 1 @@ -612,7 +611,7 @@ class TechnicalEngine: def _detect_divergences(self, df: pd.DataFrame, result: TechnicalAnalysis): """ Detect RSI and MACD divergences (powerful reversal signals). - + Bullish divergence: Price makes lower low, indicator makes higher low Bearish divergence: Price makes higher high, indicator makes lower high """ @@ -635,16 +634,15 @@ class TechnicalEngine: price_window = close.iloc[-lookback:] ind_window = indicator_series.iloc[-lookback:] - # Find local swing lows (for bullish divergence) + # Bullish divergence: lower price low but higher indicator low price_lows = self._find_swing_points(price_window, mode="low") ind_lows = self._find_swing_points(ind_window, mode="low") if len(price_lows) >= 2 and len(ind_lows) >= 2: - # Price: lower low, Indicator: higher low if price_lows[-1][1] < price_lows[-2][1] and ind_lows[-1][1] > ind_lows[-2][1]: divergences.append(f"Bullish {name} divergence (reversal warning)") - # Find local swing highs (for bearish divergence) + # Bearish divergence: higher price high but lower indicator high price_highs = self._find_swing_points(price_window, mode="high") ind_highs = self._find_swing_points(ind_window, mode="high") @@ -667,9 +665,10 @@ class TechnicalEngine: name="Divergence", value=len(divergences), signal=signal, confidence=70, description=" | ".join(divergences), - weight=1.5, # high weight — divergences are strong signals + weight=1.5, # divergences are strong reversal signals ) - result.pattern_signals.extend(divergences) + # Note: divergences are NOT added to pattern_signals to avoid + # duplicate reason collection (the indicator entry covers them). def _detect_candlestick_patterns(self, df: pd.DataFrame, result: TechnicalAnalysis): """Detect significant candlestick patterns on recent bars.""" @@ -838,7 +837,7 @@ class TechnicalEngine: def _compute_composite(self, result: TechnicalAnalysis): """ Compute the final composite signal score. - + Weighted average of all indicator signals, adjusted for: - Market regime (trend indicators weighted more in trends) - Divergence boost (divergences override in ranging markets) @@ -860,17 +859,11 @@ class TechnicalEngine: if name.startswith("_") or ind.weight == 0: continue - # Base weight from indicator weight = ind.weight - - # Regime adjustment regime_mult = active_regime_weights.get(name, 1.0) weight *= regime_mult - # Signal to score (-2 to +2) - signal_score = ind.signal.value - - # Weight by confidence + signal_score = ind.signal.value # -2 to +2 weighted_score = signal_score * weight * (ind.confidence / 100.0) total_score += weighted_score @@ -924,4 +917,4 @@ class TechnicalEngine: if all(vals[i] >= vals[i-j] for j in range(1, window+1) if not np.isnan(vals[i-j])) and \ all(vals[i] >= vals[i+j] for j in range(1, window+1) if i+j < len(vals) and not np.isnan(vals[i+j])): points.append((i, vals[i])) - return points \ No newline at end of file + return points diff --git a/main.py b/main.py index 6a58e85..d5a0c5f 100644 --- a/main.py +++ b/main.py @@ -1,503 +1,509 @@ -#!/usr/bin/env python3 -""" -Forex Signal Intelligence Platform -==================================== -Main entry point. Runs full analysis and outputs formatted signal reports. - -Usage: - python main.py # Analyze all major pairs - python main.py --pairs EURUSD GBPUSD # Specific pairs - python main.py --timeframe H4 # Different primary TF - python main.py --json # JSON output - python main.py --watch # Continuous monitoring mode -""" - -import argparse -import json -import logging -import sys -import time -from datetime import datetime, timezone -from typing import List - -from config import ( - PlatformConfig, Timeframe, SignalStrength, - MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, -) -from signal_aggregator import SignalAggregator, SignalReport - - -# ══════════════════════════════════════════════════════════════════ -# TERMINAL FORMATTING -# ══════════════════════════════════════════════════════════════════ - -class Colors: - """ANSI color codes for terminal output.""" - RESET = "\033[0m" - BOLD = "\033[1m" - DIM = "\033[2m" - UNDERLINE = "\033[4m" - - RED = "\033[91m" - GREEN = "\033[92m" - YELLOW = "\033[93m" - BLUE = "\033[94m" - MAGENTA = "\033[95m" - CYAN = "\033[96m" - WHITE = "\033[97m" - GRAY = "\033[90m" - - BG_RED = "\033[41m" - BG_GREEN = "\033[42m" - BG_YELLOW = "\033[43m" - BG_BLUE = "\033[44m" - BG_MAGENTA = "\033[45m" - BG_CYAN = "\033[46m" - - -def signal_color(signal: SignalStrength) -> str: - """Get color for a signal strength.""" - return { - SignalStrength.STRONG_BUY: Colors.GREEN + Colors.BOLD, - SignalStrength.BUY: Colors.GREEN, - SignalStrength.NEUTRAL: Colors.GRAY, - SignalStrength.SELL: Colors.RED, - SignalStrength.STRONG_SELL: Colors.RED + Colors.BOLD, - }.get(signal, Colors.RESET) - - -def quality_color(quality: str) -> str: - """Get color for signal quality grade.""" - return { - "A+": Colors.GREEN + Colors.BOLD, - "A": Colors.GREEN, - "B": Colors.CYAN, - "C": Colors.YELLOW, - "D": Colors.GRAY, - "F": Colors.DIM, - }.get(quality, Colors.RESET) - - -def score_bar(score: float, width: int = 30) -> str: - """Create a visual score bar from -100 to +100.""" - normalized = (score + 100) / 200 # 0 to 1 - filled = int(normalized * width) - center = width // 2 - - bar = "" - for i in range(width): - if i == center: - bar += "│" - elif (score >= 0 and center < i <= filled) or \ - (score < 0 and filled <= i < center): - bar += "█" - else: - bar += "░" - - if score >= 0: - color = Colors.GREEN if score > 20 else Colors.GRAY - else: - color = Colors.RED if score < -20 else Colors.GRAY - - return f"{color}{bar}{Colors.RESET}" - - -def print_header(): - """Print the platform header.""" - now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") - width = 80 - - print(f"\n{Colors.CYAN}{'═' * width}{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.CYAN} ╔═══════════════════════════════════════════════════════╗{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.CYAN} ║ FOREX SIGNAL INTELLIGENCE PLATFORM ║{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.CYAN} ║ Multi-Layer Signal Analysis Engine ║{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.CYAN} ╚═══════════════════════════════════════════════════════╝{Colors.RESET}") - print(f"{Colors.GRAY} Scan Time: {now}{Colors.RESET}") - print(f"{Colors.CYAN}{'═' * width}{Colors.RESET}\n") - - -def print_summary_table(reports: List[SignalReport]): - """Print a summary table of all pairs.""" - print(f"{Colors.BOLD}{Colors.WHITE} ┌─────────────┬───────────┬──────────────────────────────────┬────────┬────────────┐{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.WHITE} │ Pair │ Signal │ Score │ Grade │ Confidence │{Colors.RESET}") - print(f"{Colors.BOLD}{Colors.WHITE} ├─────────────┼───────────┼──────────────────────────────────┼────────┼────────────┤{Colors.RESET}") - - for r in reports: - sig = r.final_signal.name.replace("_", " ") - sc = signal_color(r.final_signal) - qc = quality_color(r.signal_quality) - bar = score_bar(r.final_score, 30) - - pair = f"{r.display_name:<11}" - signal_str = f"{sc}{sig:<9}{Colors.RESET}" - grade_str = f"{qc}{r.signal_quality:<6}{Colors.RESET}" - conf_str = f"{r.final_confidence:5.1f}%" - - # Add warning indicator - warn = "" - if r.event_risk == "danger": - warn = f" {Colors.RED}⚠{Colors.RESET}" - elif r.event_risk == "caution": - warn = f" {Colors.YELLOW}△{Colors.RESET}" - - print(f" │ {pair} │ {signal_str} │ {bar} │ {grade_str} │ {conf_str:>8}{warn} │") - - print(f"{Colors.BOLD}{Colors.WHITE} └─────────────┴───────────┴──────────────────────────────────┴────────┴────────────┘{Colors.RESET}") - - # Legend - print(f"\n{Colors.GRAY} Legend: " - f"{Colors.GREEN}██ Bullish{Colors.GRAY} | " - f"{Colors.RED}██ Bearish{Colors.GRAY} | " - f"│ Center (0) | " - f"{Colors.RED}⚠ Event Risk{Colors.GRAY} | " - f"{Colors.YELLOW}△ Caution{Colors.RESET}") - - -def print_detailed_report(report: SignalReport): - """Print detailed analysis for a single pair.""" - sc = signal_color(report.final_signal) - qc = quality_color(report.signal_quality) - sig_name = report.final_signal.name.replace("_", " ") - - print(f"\n{Colors.CYAN}{'─' * 80}{Colors.RESET}") - print(f" {Colors.BOLD}{Colors.WHITE}{report.display_name}{Colors.RESET}" - f" │ Price: {Colors.BOLD}{report.current_price:.5f}{Colors.RESET}" - f" │ Regime: {Colors.CYAN}{report.market_regime}{Colors.RESET}" - f" │ Vol: {report.volatility_regime}") - - print(f"\n {Colors.BOLD}SIGNAL:{Colors.RESET} {sc}{sig_name}{Colors.RESET}" - f" │ Score: {sc}{report.final_score:+.1f}{Colors.RESET}" - f" │ Grade: {qc}{report.signal_quality}{Colors.RESET}" - f" │ Confidence: {report.final_confidence:.0f}%") - - print(f" {Colors.BOLD}Action:{Colors.RESET} {report.action_recommendation}") - - # Multi-timeframe breakdown - if report.mtf_signals: - print(f"\n {Colors.BOLD}Multi-Timeframe:{Colors.RESET}", end="") - mtf_agree = f" {Colors.GREEN}✓ ALIGNED{Colors.RESET}" if report.mtf_agreement else f" {Colors.YELLOW}✗ DIVERGENT{Colors.RESET}" - print(mtf_agree) - for tf, sig in report.mtf_signals.items(): - tf_sc = signal_color(sig) - tf_score = report.mtf_scores.get(tf, 0) - print(f" {tf:>5}: {tf_sc}{sig.name.replace('_', ' '):<12}{Colors.RESET} ({tf_score:+.1f})") - - # Indicators detail - if report.primary_analysis: - print(f"\n {Colors.BOLD}Indicators ({report.primary_analysis.timeframe}):{Colors.RESET}") - for name, ind in report.primary_analysis.indicators.items(): - if name.startswith("_") or ind.weight == 0: - continue - isc = signal_color(ind.signal) - ind_name = f"{ind.name:<18}" - print(f" {ind_name} {isc}{ind.signal.name.replace('_', ' '):<12}{Colors.RESET}" - f" ({ind.confidence:.0f}%) │ {Colors.GRAY}{ind.description}{Colors.RESET}") - - # Bullish reasons - if report.bullish_reasons: - print(f"\n {Colors.GREEN}▲ Bullish Factors:{Colors.RESET}") - for reason in report.bullish_reasons[:5]: - print(f" {Colors.GREEN}+{Colors.RESET} {reason}") - - # Bearish reasons - if report.bearish_reasons: - print(f"\n {Colors.RED}▼ Bearish Factors:{Colors.RESET}") - for reason in report.bearish_reasons[:5]: - print(f" {Colors.RED}-{Colors.RESET} {reason}") - - # Warnings - if report.warnings: - print(f"\n {Colors.YELLOW}⚠ Warnings:{Colors.RESET}") - for warning in report.warnings: - print(f" {Colors.YELLOW}!{Colors.RESET} {warning}") - - # News context - if report.news_description and report.news_description != "Neutral news sentiment": - print(f"\n {Colors.MAGENTA}📰 News:{Colors.RESET} {report.news_description}") - if report.event_description and report.event_risk != "clear": - print(f" {Colors.YELLOW}📅 Event:{Colors.RESET} {report.event_description}") - - # Trade setup - if report.trade_setup: - ts = report.trade_setup - dir_color = Colors.GREEN if ts.direction == "BUY" else Colors.RED - print(f"\n {Colors.BOLD}Trade Setup:{Colors.RESET}") - print(f" Direction: {dir_color}{ts.direction}{Colors.RESET}") - print(f" Entry: {ts.entry_price:.5f}") - print(f" Stop Loss: {Colors.RED}{ts.stop_loss:.5f}{Colors.RESET} ({ts.risk_pips:.1f} pips)") - print(f" Take Profit 1: {Colors.GREEN}{ts.take_profit_1:.5f}{Colors.RESET} ({ts.reward_pips:.1f} pips)") - print(f" Take Profit 2: {Colors.GREEN}{ts.take_profit_2:.5f}{Colors.RESET}") - print(f" Risk:Reward: 1:{ts.risk_reward:.1f}") - - # Key levels - if report.primary_analysis and report.primary_analysis.key_levels: - levels = report.primary_analysis.key_levels - print(f"\n {Colors.BOLD}Key Levels:{Colors.RESET}") - level_names = ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"] - for lvl in level_names: - if lvl in levels: - val = levels[lvl] - if val > report.current_price: - color = Colors.RED # resistance - else: - color = Colors.GREEN # support - label = lvl.upper() - print(f" {label:>6}: {color}{val:.5f}{Colors.RESET}") - - -def print_event_calendar(reports: List[SignalReport]): - """Print upcoming economic events affecting watched pairs.""" - events_shown = set() - has_events = False - - for report in reports: - if report.event_risk != "clear" and report.event_description not in events_shown: - if not has_events: - print(f"\n{Colors.YELLOW}{Colors.BOLD} 📅 ECONOMIC CALENDAR ALERTS{Colors.RESET}") - has_events = True - events_shown.add(report.event_description) - risk_color = Colors.RED if report.event_risk == "danger" else Colors.YELLOW - print(f" {risk_color}● {report.event_description}{Colors.RESET} → affects {report.display_name}") - - if not has_events: - print(f"\n{Colors.GRAY} 📅 No high-impact events in the near term{Colors.RESET}") - - -def report_to_dict(report: SignalReport) -> dict: - """Convert a SignalReport to a JSON-serializable dict.""" - d = { - "symbol": report.symbol, - "display_name": report.display_name, - "timestamp": report.timestamp, - "current_price": report.current_price, - "signal": report.final_signal.name, - "score": round(report.final_score, 2), - "confidence": round(report.final_confidence, 2), - "quality": report.signal_quality, - "action": report.action_recommendation, - "market_regime": report.market_regime, - "volatility_regime": report.volatility_regime, - "mtf_agreement": report.mtf_agreement, - "mtf_signals": {k: v.name for k, v in report.mtf_signals.items()}, - "event_risk": report.event_risk, - "event_description": report.event_description, - "news_sentiment": round(report.news_sentiment, 3), - "bullish_reasons": report.bullish_reasons, - "bearish_reasons": report.bearish_reasons, - "warnings": report.warnings, - } - if report.trade_setup: - ts = report.trade_setup - d["trade_setup"] = { - "direction": ts.direction, - "entry": ts.entry_price, - "stop_loss": ts.stop_loss, - "take_profit_1": ts.take_profit_1, - "take_profit_2": ts.take_profit_2, - "risk_reward": ts.risk_reward, - "risk_pips": round(ts.risk_pips, 1), - "reward_pips": round(ts.reward_pips, 1), - } - if report.primary_analysis: - d["indicators"] = {} - for name, ind in report.primary_analysis.indicators.items(): - if not name.startswith("_"): - d["indicators"][name] = { - "signal": ind.signal.name, - "confidence": round(ind.confidence, 1), - "description": ind.description, - } - d["key_levels"] = { - k: round(v, 5) for k, v in report.primary_analysis.key_levels.items() - } - return d - - -# ══════════════════════════════════════════════════════════════════ -# MAIN -# ══════════════════════════════════════════════════════════════════ - -def parse_args(): - parser = argparse.ArgumentParser( - description="Forex Signal Intelligence Platform", - formatter_class=argparse.RawDescriptionHelpFormatter, - epilog=""" -Examples: - python main.py Analyze all major pairs (H1) - python main.py --pairs EURUSD GBPUSD Analyze specific pairs - python main.py --timeframe H4 Use 4-hour as primary TF - python main.py --all Include cross pairs - python main.py --json Output as JSON - python main.py --detail Show detailed analysis per pair - python main.py --watch Continuous monitoring (5min refresh) - """, - ) - parser.add_argument( - "--pairs", nargs="+", default=None, - help="Specific pairs to analyze (e.g., EURUSD GBPUSD)", - ) - parser.add_argument( - "--timeframe", "-tf", default="H1", - choices=["M5", "M15", "M30", "H1", "H4", "D1"], - help="Primary analysis timeframe (default: H1)", - ) - parser.add_argument( - "--all", action="store_true", - help="Include cross pairs in analysis", - ) - parser.add_argument( - "--json", action="store_true", - help="Output results as JSON", - ) - parser.add_argument( - "--detail", action="store_true", - help="Show detailed analysis for each pair", - ) - parser.add_argument( - "--top", type=int, default=None, - help="Show only top N signals by strength", - ) - parser.add_argument( - "--watch", action="store_true", - help="Continuous monitoring mode (refreshes every 5 min)", - ) - parser.add_argument( - "--min-grade", default="F", - choices=["A+", "A", "B", "C", "D", "F"], - help="Minimum signal quality to display", - ) - parser.add_argument( - "--verbose", "-v", action="store_true", - help="Enable debug logging", - ) - return parser.parse_args() - - -def symbol_from_name(name: str) -> str: - """Convert pair name like 'EURUSD' to Yahoo symbol 'EURUSD=X'.""" - name = name.upper().replace("/", "").replace("_", "") - if not name.endswith("=X"): - name += "=X" - return name - - -def run_analysis(args) -> List[SignalReport]: - """Run a single analysis cycle.""" - config = PlatformConfig() - - # Configure timeframe - tf_map = { - "M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30, - "H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1, - } - config.primary_timeframe = tf_map[args.timeframe] - - # Configure confirmation timeframes based on primary - tf_confirmations = { - "M5": [Timeframe.M1, Timeframe.M15], - "M15": [Timeframe.M5, Timeframe.H1], - "M30": [Timeframe.M15, Timeframe.H1], - "H1": [Timeframe.M15, Timeframe.H4], - "H4": [Timeframe.H1, Timeframe.D1], - "D1": [Timeframe.H4, Timeframe.W1], - } - config.confirmation_timeframes = tf_confirmations.get( - args.timeframe, [Timeframe.M15, Timeframe.H4] - ) - - # Configure pairs - if args.pairs: - config.watch_pairs = [symbol_from_name(p) for p in args.pairs] - elif args.all: - config.watch_pairs = MAJOR_PAIRS + CROSS_PAIRS - else: - config.watch_pairs = MAJOR_PAIRS - - # Run analysis - aggregator = SignalAggregator(config) - reports = aggregator.analyze_all_pairs() - - # Filter by grade - grade_order = ["A+", "A", "B", "C", "D", "F"] - min_idx = grade_order.index(args.min_grade) if args.min_grade in grade_order else len(grade_order) - 1 - reports = [r for r in reports if grade_order.index(r.signal_quality) <= min_idx] - - # Top N - if args.top: - reports = reports[:args.top] - - return reports - - -def main(): - args = parse_args() - - # Setup logging - log_level = logging.DEBUG if args.verbose else logging.WARNING - logging.basicConfig( - level=log_level, - format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", - datefmt="%H:%M:%S", - ) - - if args.watch: - # Continuous monitoring mode - print(f"{Colors.BOLD}Starting continuous monitoring mode (Ctrl+C to stop)...{Colors.RESET}") - while True: - try: - reports = run_analysis(args) - if args.json: - output = [report_to_dict(r) for r in reports] - print(json.dumps(output, indent=2)) - else: - print("\033[2J\033[H") # clear screen - print_header() - print_summary_table(reports) - print_event_calendar(reports) - if args.detail: - for r in reports: - print_detailed_report(r) - print(f"\n{Colors.GRAY} Next refresh in 5 minutes...{Colors.RESET}") - - time.sleep(300) - except KeyboardInterrupt: - print(f"\n{Colors.YELLOW}Monitoring stopped.{Colors.RESET}") - break - else: - # Single run - print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") - reports = run_analysis(args) - - if args.json: - output = [report_to_dict(r) for r in reports] - print(json.dumps(output, indent=2)) - else: - print_header() - if reports: - print_summary_table(reports) - print_event_calendar(reports) - - if args.detail: - for r in reports: - print_detailed_report(r) - else: - # Show detail for top signals only - actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] - if actionable: - print(f"\n{Colors.BOLD}{Colors.WHITE} ═══ ACTIONABLE SIGNALS (Grade B or above) ═══{Colors.RESET}") - for r in actionable: - print_detailed_report(r) - else: - print(f"\n{Colors.GRAY} No high-grade signals detected. " - f"Use --detail for full breakdown or --min-grade C to see more.{Colors.RESET}") - else: - print(f"{Colors.YELLOW} No data available. Check your internet connection.{Colors.RESET}") - - print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") - print(f"{Colors.GRAY} Disclaimer: This tool is for educational/research purposes only.{Colors.RESET}") - print(f"{Colors.GRAY} Past performance ≠ future results. Always use proper risk management.{Colors.RESET}") - print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") - - -if __name__ == "__main__": +#!/usr/bin/env python3 +""" +Forex Signal Intelligence Platform +==================================== +Main entry point. Runs full analysis and outputs formatted signal reports. + +Usage: + python main.py # Analyze all major pairs + python main.py --pairs EURUSD GBPUSD # Specific pairs + python main.py --timeframe H4 # Different primary TF + python main.py --json # JSON output + python main.py --watch # Continuous monitoring mode +""" + +import argparse +import json +import logging +import sys +import time +from datetime import datetime, timezone +from typing import List + +# Force UTF-8 output on Windows so box-drawing / Unicode chars render correctly +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") +if hasattr(sys.stderr, "reconfigure"): + sys.stderr.reconfigure(encoding="utf-8") + +from config import ( + PlatformConfig, Timeframe, SignalStrength, + MAJOR_PAIRS, CROSS_PAIRS, +) +from signal_aggregator import SignalAggregator, SignalReport + + +# ══════════════════════════════════════════════════════════════════ +# TERMINAL FORMATTING +# ══════════════════════════════════════════════════════════════════ + +class Colors: + """ANSI color codes for terminal output.""" + RESET = "\033[0m" + BOLD = "\033[1m" + DIM = "\033[2m" + UNDERLINE = "\033[4m" + + RED = "\033[91m" + GREEN = "\033[92m" + YELLOW = "\033[93m" + BLUE = "\033[94m" + MAGENTA = "\033[95m" + CYAN = "\033[96m" + WHITE = "\033[97m" + GRAY = "\033[90m" + + BG_RED = "\033[41m" + BG_GREEN = "\033[42m" + BG_YELLOW = "\033[43m" + BG_BLUE = "\033[44m" + BG_MAGENTA = "\033[45m" + BG_CYAN = "\033[46m" + + +def signal_color(signal: SignalStrength) -> str: + """Get color for a signal strength.""" + return { + SignalStrength.STRONG_BUY: Colors.GREEN + Colors.BOLD, + SignalStrength.BUY: Colors.GREEN, + SignalStrength.NEUTRAL: Colors.GRAY, + SignalStrength.SELL: Colors.RED, + SignalStrength.STRONG_SELL: Colors.RED + Colors.BOLD, + }.get(signal, Colors.RESET) + + +def quality_color(quality: str) -> str: + """Get color for signal quality grade.""" + return { + "A+": Colors.GREEN + Colors.BOLD, + "A": Colors.GREEN, + "B": Colors.CYAN, + "C": Colors.YELLOW, + "D": Colors.GRAY, + "F": Colors.DIM, + }.get(quality, Colors.RESET) + + +def score_bar(score: float, width: int = 30) -> str: + """Create a visual score bar from -100 to +100.""" + normalized = (score + 100) / 200 # 0 to 1 + filled = int(normalized * width) + center = width // 2 + + bar = "" + for i in range(width): + if i == center: + bar += "│" + elif (score >= 0 and center < i <= filled) or \ + (score < 0 and filled <= i < center): + bar += "█" + else: + bar += "░" + + if score >= 0: + color = Colors.GREEN if score > 20 else Colors.GRAY + else: + color = Colors.RED if score < -20 else Colors.GRAY + + return f"{color}{bar}{Colors.RESET}" + + +def print_header(): + """Print the platform header.""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + width = 80 + + print(f"\n{Colors.CYAN}{'═' * width}{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ╔═══════════════════════════════════════════════════════╗{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ║ FOREX SIGNAL INTELLIGENCE PLATFORM ║{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ║ Multi-Layer Signal Analysis Engine ║{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ╚═══════════════════════════════════════════════════════╝{Colors.RESET}") + print(f"{Colors.GRAY} Scan Time: {now}{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * width}{Colors.RESET}\n") + + +def print_summary_table(reports: List[SignalReport]): + """Print a summary table of all pairs.""" + print(f"{Colors.BOLD}{Colors.WHITE} ┌─────────────┬───────────┬──────────────────────────────────┬────────┬────────────┐{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.WHITE} │ Pair │ Signal │ Score │ Grade │ Confidence │{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.WHITE} ├─────────────┼───────────┼──────────────────────────────────┼────────┼────────────┤{Colors.RESET}") + + for r in reports: + sig = r.final_signal.name.replace("_", " ") + sc = signal_color(r.final_signal) + qc = quality_color(r.signal_quality) + bar = score_bar(r.final_score, 30) + + pair = f"{r.display_name:<11}" + signal_str = f"{sc}{sig:<9}{Colors.RESET}" + grade_str = f"{qc}{r.signal_quality:<6}{Colors.RESET}" + conf_str = f"{r.final_confidence:5.1f}%" + + # Add warning indicator + warn = "" + if r.event_risk == "danger": + warn = f" {Colors.RED}⚠{Colors.RESET}" + elif r.event_risk == "caution": + warn = f" {Colors.YELLOW}△{Colors.RESET}" + + print(f" │ {pair} │ {signal_str} │ {bar} │ {grade_str} │ {conf_str:>8}{warn} │") + + print(f"{Colors.BOLD}{Colors.WHITE} └─────────────┴───────────┴──────────────────────────────────┴────────┴────────────┘{Colors.RESET}") + + # Legend + print(f"\n{Colors.GRAY} Legend: " + f"{Colors.GREEN}██ Bullish{Colors.GRAY} | " + f"{Colors.RED}██ Bearish{Colors.GRAY} | " + f"│ Center (0) | " + f"{Colors.RED}⚠ Event Risk{Colors.GRAY} | " + f"{Colors.YELLOW}△ Caution{Colors.RESET}") + + +def print_detailed_report(report: SignalReport): + """Print detailed analysis for a single pair.""" + sc = signal_color(report.final_signal) + qc = quality_color(report.signal_quality) + sig_name = report.final_signal.name.replace("_", " ") + + print(f"\n{Colors.CYAN}{'─' * 80}{Colors.RESET}") + print(f" {Colors.BOLD}{Colors.WHITE}{report.display_name}{Colors.RESET}" + f" │ Price: {Colors.BOLD}{report.current_price:.5f}{Colors.RESET}" + f" │ Regime: {Colors.CYAN}{report.market_regime}{Colors.RESET}" + f" │ Vol: {report.volatility_regime}") + + print(f"\n {Colors.BOLD}SIGNAL:{Colors.RESET} {sc}{sig_name}{Colors.RESET}" + f" │ Score: {sc}{report.final_score:+.1f}{Colors.RESET}" + f" │ Grade: {qc}{report.signal_quality}{Colors.RESET}" + f" │ Confidence: {report.final_confidence:.0f}%") + + print(f" {Colors.BOLD}Action:{Colors.RESET} {report.action_recommendation}") + + # Multi-timeframe breakdown + if report.mtf_signals: + print(f"\n {Colors.BOLD}Multi-Timeframe:{Colors.RESET}", end="") + mtf_agree = f" {Colors.GREEN}✓ ALIGNED{Colors.RESET}" if report.mtf_agreement else f" {Colors.YELLOW}✗ DIVERGENT{Colors.RESET}" + print(mtf_agree) + for tf, sig in report.mtf_signals.items(): + tf_sc = signal_color(sig) + tf_score = report.mtf_scores.get(tf, 0) + print(f" {tf:>5}: {tf_sc}{sig.name.replace('_', ' '):<12}{Colors.RESET} ({tf_score:+.1f})") + + # Indicators detail + if report.primary_analysis: + print(f"\n {Colors.BOLD}Indicators ({report.primary_analysis.timeframe}):{Colors.RESET}") + for name, ind in report.primary_analysis.indicators.items(): + if name.startswith("_") or ind.weight == 0: + continue + isc = signal_color(ind.signal) + ind_name = f"{ind.name:<18}" + print(f" {ind_name} {isc}{ind.signal.name.replace('_', ' '):<12}{Colors.RESET}" + f" ({ind.confidence:.0f}%) │ {Colors.GRAY}{ind.description}{Colors.RESET}") + + # Bullish reasons + if report.bullish_reasons: + print(f"\n {Colors.GREEN}▲ Bullish Factors:{Colors.RESET}") + for reason in report.bullish_reasons[:5]: + print(f" {Colors.GREEN}+{Colors.RESET} {reason}") + + # Bearish reasons + if report.bearish_reasons: + print(f"\n {Colors.RED}▼ Bearish Factors:{Colors.RESET}") + for reason in report.bearish_reasons[:5]: + print(f" {Colors.RED}-{Colors.RESET} {reason}") + + # Warnings + if report.warnings: + print(f"\n {Colors.YELLOW}⚠ Warnings:{Colors.RESET}") + for warning in report.warnings: + print(f" {Colors.YELLOW}!{Colors.RESET} {warning}") + + # News context + if report.news_description and report.news_description != "Neutral news sentiment": + print(f"\n {Colors.MAGENTA}📰 News:{Colors.RESET} {report.news_description}") + if report.event_description and report.event_risk != "clear": + print(f" {Colors.YELLOW}📅 Event:{Colors.RESET} {report.event_description}") + + # Trade setup + if report.trade_setup: + ts = report.trade_setup + dir_color = Colors.GREEN if ts.direction == "BUY" else Colors.RED + print(f"\n {Colors.BOLD}Trade Setup:{Colors.RESET}") + print(f" Direction: {dir_color}{ts.direction}{Colors.RESET}") + print(f" Entry: {ts.entry_price:.5f}") + print(f" Stop Loss: {Colors.RED}{ts.stop_loss:.5f}{Colors.RESET} ({ts.risk_pips:.1f} pips)") + print(f" Take Profit 1: {Colors.GREEN}{ts.take_profit_1:.5f}{Colors.RESET} ({ts.reward_pips:.1f} pips)") + print(f" Take Profit 2: {Colors.GREEN}{ts.take_profit_2:.5f}{Colors.RESET}") + print(f" Risk:Reward: 1:{ts.risk_reward:.1f}") + + # Key levels + if report.primary_analysis and report.primary_analysis.key_levels: + levels = report.primary_analysis.key_levels + print(f"\n {Colors.BOLD}Key Levels:{Colors.RESET}") + level_names = ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"] + for lvl in level_names: + if lvl in levels: + val = levels[lvl] + if val > report.current_price: + color = Colors.RED # resistance + else: + color = Colors.GREEN # support + label = lvl.upper() + print(f" {label:>6}: {color}{val:.5f}{Colors.RESET}") + + +def print_event_calendar(reports: List[SignalReport]): + """Print upcoming economic events affecting watched pairs.""" + events_shown = set() + has_events = False + + for report in reports: + if report.event_risk != "clear" and report.event_description not in events_shown: + if not has_events: + print(f"\n{Colors.YELLOW}{Colors.BOLD} 📅 ECONOMIC CALENDAR ALERTS{Colors.RESET}") + has_events = True + events_shown.add(report.event_description) + risk_color = Colors.RED if report.event_risk == "danger" else Colors.YELLOW + print(f" {risk_color}● {report.event_description}{Colors.RESET} → affects {report.display_name}") + + if not has_events: + print(f"\n{Colors.GRAY} 📅 No high-impact events in the near term{Colors.RESET}") + + +def report_to_dict(report: SignalReport) -> dict: + """Convert a SignalReport to a JSON-serializable dict.""" + d = { + "symbol": report.symbol, + "display_name": report.display_name, + "timestamp": report.timestamp, + "current_price": report.current_price, + "signal": report.final_signal.name, + "score": round(report.final_score, 2), + "confidence": round(report.final_confidence, 2), + "quality": report.signal_quality, + "action": report.action_recommendation, + "market_regime": report.market_regime, + "volatility_regime": report.volatility_regime, + "mtf_agreement": report.mtf_agreement, + "mtf_signals": {k: v.name for k, v in report.mtf_signals.items()}, + "event_risk": report.event_risk, + "event_description": report.event_description, + "news_sentiment": round(report.news_sentiment, 3), + "bullish_reasons": report.bullish_reasons, + "bearish_reasons": report.bearish_reasons, + "warnings": report.warnings, + } + if report.trade_setup: + ts = report.trade_setup + d["trade_setup"] = { + "direction": ts.direction, + "entry": ts.entry_price, + "stop_loss": ts.stop_loss, + "take_profit_1": ts.take_profit_1, + "take_profit_2": ts.take_profit_2, + "risk_reward": ts.risk_reward, + "risk_pips": round(ts.risk_pips, 1), + "reward_pips": round(ts.reward_pips, 1), + } + if report.primary_analysis: + d["indicators"] = {} + for name, ind in report.primary_analysis.indicators.items(): + if not name.startswith("_"): + d["indicators"][name] = { + "signal": ind.signal.name, + "confidence": round(ind.confidence, 1), + "description": ind.description, + } + d["key_levels"] = { + k: round(v, 5) for k, v in report.primary_analysis.key_levels.items() + } + return d + + +# ══════════════════════════════════════════════════════════════════ +# MAIN +# ══════════════════════════════════════════════════════════════════ + +def parse_args(): + parser = argparse.ArgumentParser( + description="Forex Signal Intelligence Platform", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + python main.py Analyze all major pairs (H1) + python main.py --pairs EURUSD GBPUSD Analyze specific pairs + python main.py --timeframe H4 Use 4-hour as primary TF + python main.py --all Include cross pairs + python main.py --json Output as JSON + python main.py --detail Show detailed analysis per pair + python main.py --watch Continuous monitoring (5min refresh) + """, + ) + parser.add_argument( + "--pairs", nargs="+", default=None, + help="Specific pairs to analyze (e.g., EURUSD GBPUSD)", + ) + parser.add_argument( + "--timeframe", "-tf", default="H1", + choices=["M5", "M15", "M30", "H1", "H4", "D1"], + help="Primary analysis timeframe (default: H1)", + ) + parser.add_argument( + "--all", action="store_true", + help="Include cross pairs in analysis", + ) + parser.add_argument( + "--json", action="store_true", + help="Output results as JSON", + ) + parser.add_argument( + "--detail", action="store_true", + help="Show detailed analysis for each pair", + ) + parser.add_argument( + "--top", type=int, default=None, + help="Show only top N signals by strength", + ) + parser.add_argument( + "--watch", action="store_true", + help="Continuous monitoring mode (refreshes every 5 min)", + ) + parser.add_argument( + "--min-grade", default="F", + choices=["A+", "A", "B", "C", "D", "F"], + help="Minimum signal quality to display", + ) + parser.add_argument( + "--verbose", "-v", action="store_true", + help="Enable debug logging", + ) + return parser.parse_args() + + +def symbol_from_name(name: str) -> str: + """Convert pair name like 'EURUSD' to Yahoo symbol 'EURUSD=X'.""" + name = name.upper().replace("/", "").replace("_", "") + if not name.endswith("=X"): + name += "=X" + return name + + +def run_analysis(args) -> List[SignalReport]: + """Run a single analysis cycle.""" + config = PlatformConfig() + + # Configure timeframe + tf_map = { + "M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30, + "H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1, + } + config.primary_timeframe = tf_map[args.timeframe] + + # Configure confirmation timeframes based on primary + tf_confirmations = { + "M5": [Timeframe.M1, Timeframe.M15], + "M15": [Timeframe.M5, Timeframe.H1], + "M30": [Timeframe.M15, Timeframe.H1], + "H1": [Timeframe.M15, Timeframe.H4], + "H4": [Timeframe.H1, Timeframe.D1], + "D1": [Timeframe.H4, Timeframe.W1], + } + config.confirmation_timeframes = tf_confirmations.get( + args.timeframe, [Timeframe.M15, Timeframe.H4] + ) + + # Configure pairs + if args.pairs: + config.watch_pairs = [symbol_from_name(p) for p in args.pairs] + elif args.all: + config.watch_pairs = MAJOR_PAIRS + CROSS_PAIRS + else: + config.watch_pairs = MAJOR_PAIRS + + # Run analysis + aggregator = SignalAggregator(config) + reports = aggregator.analyze_all_pairs() + + # Filter by grade + grade_order = ["A+", "A", "B", "C", "D", "F"] + min_idx = grade_order.index(args.min_grade) if args.min_grade in grade_order else len(grade_order) - 1 + reports = [r for r in reports if grade_order.index(r.signal_quality) <= min_idx] + + # Top N + if args.top: + reports = reports[:args.top] + + return reports + + +def main(): + args = parse_args() + + # Setup logging + log_level = logging.DEBUG if args.verbose else logging.WARNING + logging.basicConfig( + level=log_level, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%H:%M:%S", + ) + + if args.watch: + # Continuous monitoring mode + print(f"{Colors.BOLD}Starting continuous monitoring mode (Ctrl+C to stop)...{Colors.RESET}") + while True: + try: + reports = run_analysis(args) + if args.json: + output = [report_to_dict(r) for r in reports] + print(json.dumps(output, indent=2)) + else: + print("\033[2J\033[H") # clear screen + print_header() + print_summary_table(reports) + print_event_calendar(reports) + if args.detail: + for r in reports: + print_detailed_report(r) + print(f"\n{Colors.GRAY} Next refresh in 5 minutes...{Colors.RESET}") + + time.sleep(300) + except KeyboardInterrupt: + print(f"\n{Colors.YELLOW}Monitoring stopped.{Colors.RESET}") + break + else: + # Single run + print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") + reports = run_analysis(args) + + if args.json: + output = [report_to_dict(r) for r in reports] + print(json.dumps(output, indent=2)) + else: + print_header() + if reports: + print_summary_table(reports) + print_event_calendar(reports) + + if args.detail: + for r in reports: + print_detailed_report(r) + else: + # Show detail for top signals only + actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] + if actionable: + print(f"\n{Colors.BOLD}{Colors.WHITE} ═══ ACTIONABLE SIGNALS (Grade B or above) ═══{Colors.RESET}") + for r in actionable: + print_detailed_report(r) + else: + print(f"\n{Colors.GRAY} No high-grade signals detected. " + f"Use --detail for full breakdown or --min-grade C to see more.{Colors.RESET}") + else: + print(f"{Colors.YELLOW} No data available. Check your internet connection.{Colors.RESET}") + + print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") + print(f"{Colors.GRAY} Disclaimer: This tool is for educational/research purposes only.{Colors.RESET}") + print(f"{Colors.GRAY} Past performance ≠ future results. Always use proper risk management.{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") + + +if __name__ == "__main__": main() \ No newline at end of file diff --git a/news_engine.py b/news_engine.py index b254a3c..7424c62 100644 --- a/news_engine.py +++ b/news_engine.py @@ -9,7 +9,6 @@ A perfect technical setup can be destroyed by an unexpected NFP print. """ import logging -import re import xml.etree.ElementTree as ET from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone @@ -69,12 +68,13 @@ class FundamentalContext: news_sentiment: Dict[str, float] = field(default_factory=dict) # currency -> sentiment news_headlines: List[NewsSentiment] = field(default_factory=list) event_impact_on_pairs: Dict[str, str] = field(default_factory=dict) + calendar_available: bool = False # True if live calendar data was fetched class NewsEngine: """ Economic calendar and news sentiment analyzer. - + Provides fundamental context that modifies technical signal confidence. """ @@ -126,7 +126,7 @@ class NewsEngine: ) -> FundamentalContext: """ Get complete fundamental context for signal decisions. - + Combines economic calendar + news sentiment into actionable context. """ if currencies is None: @@ -144,6 +144,7 @@ class NewsEngine: e for e in events if 0 < e.minutes_until <= 60 ] + context.calendar_available = bool(events) # Check for high-impact warnings for event in events: @@ -155,7 +156,7 @@ class NewsEngine: ) break - # Map events to affected pairs + # Map upcoming high-impact events to affected pairs for event in context.events_upcoming_1h: if event.is_high_impact: affected = self._get_affected_pairs(event.currency) @@ -192,7 +193,6 @@ class NewsEngine: Returns (risk_level, description). risk_level: "clear", "caution", "danger" """ - # Extract currencies from pair clean = pair_symbol.replace("=X", "") base = clean[:3] quote = clean[3:] @@ -237,11 +237,11 @@ class NewsEngine: elif net > 0.3: desc = f"Bullish sentiment ({base} positive / {quote} negative)" elif net > 0: - desc = f"Slightly bullish sentiment" + desc = "Slightly bullish sentiment" elif net < -0.3: desc = f"Bearish sentiment ({base} negative / {quote} positive)" else: - desc = f"Slightly bearish sentiment" + desc = "Slightly bearish sentiment" return net, desc @@ -249,104 +249,54 @@ class NewsEngine: def _fetch_economic_calendar(self, currencies: List[str]) -> List[EconomicEvent]: """ - Fetch economic calendar events. - Uses a lightweight approach since most free APIs require auth. - Falls back to manually constructed high-impact schedule. + Fetch economic calendar events from a free API. + Returns an empty list (with a log warning) if unavailable — + we never generate fake events since that causes constant false alarms. """ - events = [] - - # Try fetching from a free calendar source try: events = self._try_fetch_calendar_api() if events: return [e for e in events if e.currency in currencies] - except Exception: - pass + except Exception as e: + logger.debug(f"Calendar API fetch exception: {e}") - # Fallback: generate known recurring high-impact events - events = self._generate_known_events(currencies) - return events + logger.info( + "Economic calendar data unavailable. " + "Event risk checks will be skipped for this cycle." + ) + return [] def _try_fetch_calendar_api(self) -> List[EconomicEvent]: - """Attempt to fetch from a free economic calendar API.""" - # Using a simple RSS/JSON approach + """Attempt to fetch from the Forex Factory free JSON calendar.""" events = [] - try: - # Try Forex Factory RSS (most widely available) - url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json" - resp = self._session.get(url, timeout=10) - if resp.status_code == 200: - data = resp.json() - for item in data: - try: - impact_map = {"High": "high", "Medium": "medium", "Low": "low"} - impact = impact_map.get(item.get("impact", ""), "low") + url = "https://nfs.faireconomy.media/ff_calendar_thisweek.json" + resp = self._session.get(url, timeout=10) + if resp.status_code != 200: + return [] - # Parse date - date_str = item.get("date", "") - if date_str: - event_time = datetime.fromisoformat( - date_str.replace("Z", "+00:00") - ) - else: - continue + data = resp.json() + impact_map = {"High": "high", "Medium": "medium", "Low": "low"} - events.append(EconomicEvent( - time=event_time, - currency=item.get("country", ""), - impact=impact, - event=item.get("title", ""), - forecast=item.get("forecast", None), - previous=item.get("previous", None), - actual=item.get("actual") if item.get("actual") else None, - )) - except Exception: - continue - except Exception as e: - logger.debug(f"Calendar API unavailable: {e}") + for item in data: + try: + impact = impact_map.get(item.get("impact", ""), "low") - return events + date_str = item.get("date", "") + if not date_str: + continue + event_time = datetime.fromisoformat(date_str.replace("Z", "+00:00")) - def _generate_known_events(self, currencies: List[str]) -> List[EconomicEvent]: - """ - Generate a schedule of known recurring high-impact events. - This is the fallback when no API is available. - """ - events = [] - now = datetime.now(timezone.utc) - today = now.date() - - # Known high-impact regular events (approximate times) - recurring = [ - ("USD", "FOMC Rate Decision", 18, 0), # 2pm ET - ("USD", "Non-Farm Payrolls", 12, 30), # 8:30am ET (first Friday) - ("USD", "CPI Release", 12, 30), - ("EUR", "ECB Rate Decision", 12, 45), - ("GBP", "BOE Rate Decision", 11, 0), - ("AUD", "RBA Rate Decision", 3, 30), - ("CAD", "BOC Rate Decision", 14, 0), - ("JPY", "BOJ Rate Decision", 3, 0), - ("USD", "Initial Jobless Claims", 12, 30), # Weekly Thursday - ("USD", "Retail Sales", 12, 30), - ("EUR", "German ZEW", 9, 0), - ] - - # Only include if they'd be today or tomorrow - for currency, event_name, hour, minute in recurring: - if currency not in currencies: - continue - for day_offset in range(2): - event_date = today + timedelta(days=day_offset) - event_time = datetime( - event_date.year, event_date.month, event_date.day, - hour, minute, tzinfo=timezone.utc, - ) events.append(EconomicEvent( time=event_time, - currency=currency, - impact="high", - event=event_name, + currency=item.get("country", ""), + impact=impact, + event=item.get("title", ""), + forecast=item.get("forecast") or None, + previous=item.get("previous") or None, + actual=item.get("actual") or None, )) + except Exception: + continue return events @@ -398,25 +348,17 @@ class NewsEngine: def _analyze_headline_sentiment(self, headline: str) -> float: """ - Simple but effective keyword-based sentiment analysis. - + Simple keyword-based sentiment analysis. + Returns -1.0 (very bearish) to +1.0 (very bullish). - + Note: This is intentionally simple. NLP models like FinBERT - would be better but require large dependencies. This catches + would be more accurate but require large dependencies. This catches the most impactful sentiment signals from headlines. """ headline_lower = headline.lower() - bullish_score = 0 - bearish_score = 0 - - for keyword in self.BULLISH_KEYWORDS: - if keyword in headline_lower: - bullish_score += 1 - - for keyword in self.BEARISH_KEYWORDS: - if keyword in headline_lower: - bearish_score += 1 + bullish_score = sum(1 for kw in self.BULLISH_KEYWORDS if kw in headline_lower) + bearish_score = sum(1 for kw in self.BEARISH_KEYWORDS if kw in headline_lower) total = bullish_score + bearish_score if total == 0: @@ -443,29 +385,28 @@ class NewsEngine: def _extract_keywords(self, headline: str) -> List[str]: """Extract relevant trading keywords from headline.""" headline_lower = headline.lower() - found = [] - for kw in self.BULLISH_KEYWORDS | self.BEARISH_KEYWORDS: - if kw in headline_lower: - found.append(kw) - return found + return [kw for kw in self.BULLISH_KEYWORDS | self.BEARISH_KEYWORDS if kw in headline_lower] @staticmethod def _parse_rss_date(date_str: Optional[str]) -> Optional[datetime]: - """Parse various RSS date formats.""" + """Parse various RSS date formats into a timezone-aware datetime.""" if not date_str: return None formats = [ - "%a, %d %b %Y %H:%M:%S %z", - "%a, %d %b %Y %H:%M:%S %Z", - "%Y-%m-%dT%H:%M:%S%z", - "%Y-%m-%d %H:%M:%S", + "%a, %d %b %Y %H:%M:%S %z", # RFC 822 with numeric tz offset + "%a, %d %b %Y %H:%M:%S %Z", # RFC 822 with tz abbreviation (GMT/UTC) + "%Y-%m-%dT%H:%M:%S%z", # ISO 8601 with offset + "%Y-%m-%d %H:%M:%S", # ISO 8601 naive (assume UTC) ] for fmt in formats: try: - return datetime.strptime(date_str.strip(), fmt).replace( - tzinfo=timezone.utc if fmt[-2:] != "%z" else None - ) + dt = datetime.strptime(date_str.strip(), fmt) + # If strptime already attached tzinfo (from %z), keep it. + # Otherwise the string had no timezone info — assume UTC. + if dt.tzinfo is not None: + return dt + return dt.replace(tzinfo=timezone.utc) except ValueError: continue return None @@ -483,4 +424,4 @@ class NewsEngine: "CAD": ["USDCAD=X", "CADJPY=X"], "NZD": ["NZDUSD=X"], } - return pairs.get(currency, []) \ No newline at end of file + return pairs.get(currency, []) diff --git a/script.txt b/script.txt index ceee9c1..cfa83a2 100644 --- a/script.txt +++ b/script.txt @@ -1 +1 @@ -pip install pandas numpy requests ta python-dateutil --break-system-packages -q 2>&1 | tail -5 \ No newline at end of file +pip install pandas numpy requests python-dateutil -q