diff --git a/USAGE.md b/USAGE.md index ff24555..8aa2b2b 100644 --- a/USAGE.md +++ b/USAGE.md @@ -1,4 +1,8 @@ -# ForexBot2 — Usage Guide +# ForexBot2 — USD/JPY Signal Intelligence + +A focused USD/JPY signal generator + walk-forward backtester. Live signals +and backtests run through the **same** technical engine, so what you measure +in a backtest is exactly what you'd execute live. ## Setup @@ -6,159 +10,160 @@ pip install -r requirements.txt ``` -Set your Twelve Data API key (or it falls back to the key in `config.py`): +API key (optional — a default key is in `config.py`): ```bash -export TWELVE_DATA_API_KEY=your_key_here # Linux/Mac -set TWELVE_DATA_API_KEY=your_key_here # Windows CMD -$env:TWELVE_DATA_API_KEY="your_key_here" # PowerShell +export TWELVE_DATA_API_KEY=your_key_here # Linux/Mac +$env:TWELVE_DATA_API_KEY="your_key_here" # PowerShell ``` --- -## Basic Syntax - -``` -python main.py [OPTIONS] -``` - ---- - -## Options - -| Option | Default | Description | -|---|---|---| -| `--pairs PAIR [PAIR ...]` | all major pairs | Specific pairs to analyze | -| `--timeframe TF` | `H1` | Primary timeframe: `M5 M15 M30 H1 H4 D1` | -| `--all` | off | Add EUR/GBP, EUR/JPY, GBP/JPY and other cross pairs | -| `--detail` | off | Full indicator breakdown per pair | -| `--top N` | all | Show only the top N signals by strength | -| `--min-grade GRADE` | `F` (show all) | Filter by minimum quality grade: `A+ A B C D F` | -| `--json` | off | Output results as JSON instead of the terminal UI | -| `--watch` | off | Continuous monitoring — refreshes every 5 minutes | -| `--verbose` / `-v` | off | Enable debug logging | - ---- - -## Common Examples - -### Analyze all major Forex pairs (default) -```bash -python main.py -``` -Analyzes EUR/USD, GBP/USD, USD/JPY, USD/CHF, AUD/USD, USD/CAD, NZD/USD on H1. - ---- - -### Analyze Gold (XAU/USD) -```bash -python main.py --pairs XAUUSD -``` -Pass the symbol without the slash. The engine converts it to Twelve Data format automatically. - ---- - -### Gold on the 4-hour chart with full detail -```bash -python main.py --pairs XAUUSD --timeframe H4 --detail -``` - ---- - -### Multiple pairs at once -```bash -python main.py --pairs XAUUSD EURUSD GBPUSD -``` - ---- - -### Only show high-conviction signals (Grade B or above) -```bash -python main.py --min-grade B -``` - ---- - -### Top 3 signals across all pairs including crosses -```bash -python main.py --all --top 3 -``` - ---- - -### Continuous monitoring — updates every 5 minutes -```bash -python main.py --watch -``` -Add `--pairs XAUUSD` to watch a single instrument, or `--all` for everything. +## Live Signal ```bash -python main.py --pairs XAUUSD --watch --detail +python main.py # H1, summary view +python main.py --detail # Full indicator breakdown +python main.py --timeframe H4 --detail # H4 instead of H1 +python main.py --json # JSON output (for piping) ``` ---- +| Option | Default | Description | +|---------------|---------|--------------------------------------------| +| `--timeframe` | `H1` | `M5 M15 M30 H1 H4 D1` | +| `--detail` | off | Indicator breakdown, key levels, reasons | +| `--json` | off | Machine-readable output | +| `--verbose` | off | Debug logging | -### JSON output (for piping or logging) -```bash -python main.py --pairs XAUUSD --json -python main.py --all --json > signals.json -``` - ---- - -## Signal Grades - -| Grade | Meaning | -|---|---| -| A+ | Strongest — all timeframes aligned, high confidence | -| A | Strong signal, minor divergence | -| B | Good setup, watch for confirmation | -| C | Weak — for reference only | -| D/F | Noise — best ignored | - ---- - -## Supported Symbols - -Pass any symbol without slashes. Examples: - -| You type | Pair | -|---|---| -| `XAUUSD` | Gold vs USD | -| `EURUSD` | Euro vs USD | -| `GBPUSD` | British Pound vs USD | -| `USDJPY` | USD vs Japanese Yen | -| `GBPJPY` | GBP vs Japanese Yen | -| `USDCAD` | USD vs Canadian Dollar | -| `AUDUSD` | Australian Dollar vs USD | - -Any pair supported by Twelve Data can be used. - ---- - -## Output Explained +### Output ``` -EUR/USD │ Price: 1.08423 │ Regime: trending_up │ Vol: normal +USD/JPY │ Price: 149.230 │ Regime: trending_up │ Vol: normal -SIGNAL: STRONG BUY │ Score: +62.4 │ Grade: A │ Confidence: 74% -Action: Strong buy signal with multi-timeframe alignment... +SIGNAL: BUY │ Score: +28.4 │ Grade: B │ Confidence: 62% +Action: BUY — Decent setup, wait for confirmation Multi-Timeframe: ✓ ALIGNED - M15: BUY (+28.1) - H1: STRONG BUY (+62.4) - H4: BUY (+41.0) + M15: BUY (+18.2) + H1: BUY (+28.4) + H4: STRONG BUY (+45.1) Trade Setup: Direction: BUY - Entry: 1.08423 - Stop Loss: 1.08150 (27.3 pips) - Take Profit 1: 1.08970 (54.7 pips) - Take Profit 2: 1.09517 + Entry: 149.230 + Stop Loss: 148.960 (27.0 pips) + Take Profit 1: 149.770 (54.0 pips) + Take Profit 2: 150.040 Risk:Reward: 1:2.0 ``` -- **Score** — ranges from -100 (strong sell) to +100 (strong buy) -- **Grade** — summary quality of the signal setup -- **Regime** — `trending_up`, `trending_down`, or `ranging` -- **Trade Setup** — auto-calculated entry, SL, and TP based on ATR +--- + +## Backtest (Walk-Forward) + +The backtester replays USD/JPY history bar-by-bar through the live signal +pipeline. No lookahead, no parallel implementation — what passes in +backtest passes live. + +```bash +python main.py --backtest # Full available history +python main.py --backtest --bars 2000 # Last 2000 H1 bars +python main.py --backtest --start 2024-01-01 --end 2024-12-31 +python main.py --backtest --timeframe H4 # Backtest on H4 +python main.py --backtest --min-grade A # Only A/A+ entries +python main.py --backtest --journal # Last 10 trades printed +python main.py --backtest --journal-csv trades.csv # Full journal to CSV +python main.py --backtest --regime-filter # Skip choppy/ranging bars +python main.py --backtest --regime-filter --regime-min-adx 35 +``` + +### Regime filter + +The engine is a confluence-based trend-follower — it makes money in trending +regimes and loses in chop. `--regime-filter` gates entries on H4 ADX: +signals only fire when ADX ≥ `--regime-min-adx` (default 30). Sweet spot is +ADX 30–35; tighter cuts trades faster than it improves quality. + +### Execution model + +| Aspect | Behavior | +|--------------------|-------------------------------------------------------------------| +| Entry | Next bar's open (no current-bar lookahead) | +| Stop Loss | 1.5 × ATR from entry (configurable) | +| Take Profit | RR × stop distance (default 1:2) | +| Same-bar SL & TP | Conservative: SL fills first | +| Spread cost | Half-spread on entry + half-spread on exit (full round-trip) | +| Slippage | 0.5 pips per fill (entry + exit) | +| Position size | 1% of current equity risked per trade | +| Concurrency | One open position at a time | +| News in backtest | Disabled — RSS/calendar can't be replayed without lookahead | + +### Output + +``` +Backtest Summary — USD/JPY +───────────────────────────────────────────────── +Trades: 87 (52W / 35L) +Win Rate: 59.8% +Profit Factor: 1.84 +Expectancy: +12.4 pips/trade (+12.42 per trade) +Initial Balance: $10,000.00 +Final Balance: $11,084.00 +Total Return: +10.84% +CAGR: +18.92% +Max Drawdown: -4.21% +Sharpe Ratio: 1.42 +Sortino Ratio: 2.18 +Avg Duration: 14.3 bars +Win Streak: 7 | Loss Streak: 4 +───────────────────────────────────────────────── +``` + +--- + +## Parameter Optimization (parallel) + +```bash +python main.py --backtest --optimize +``` + +Runs a 3×3×3 grid (27 combos) over `atr_sl_multiplier`, `default_rr_ratio`, +and `min_score` (signal-score threshold). Combos run in parallel across all +CPU cores via `multiprocessing.Pool`. Twelve Data is hit once — workers +share an on-disk parquet cache (`.cache/usdjpy_.parquet`). + +Output ranks combos by total return: + +``` +Top 10 Parameter Combinations (by total return): + +# Return% PF WinR Trades MaxDD% Sharpe Params +────────────────────────────────────────────────────────────────────── +1 +14.21% 2.10 62.5% 64 -3.84 1.78 atr_sl_multiplier=1.5, default_rr_ratio=2.5, min_score=25.0 +2 +11.93% 1.92 58.7% 71 -4.11 1.55 atr_sl_multiplier=1.0, default_rr_ratio=2.0, min_score=15.0 +... +``` + +Customize the grid in `config.py → BacktestConfig.param_grid`. + +--- + +## Files + +| File | Role | +|-----------------------|---------------------------------------------------| +| `main.py` | CLI entry point (live + backtest + optimize) | +| `config.py` | All parameters in one place | +| `data_provider.py` | Twelve Data fetcher + parquet cache | +| `engine.py` | Technical indicators (8 + divergence + patterns) | +| `signal_aggregator.py`| Multi-TF aggregation, grading, trade setup | +| `news_engine.py` | RSS sentiment + economic calendar (live only) | +| `backtest.py` | Walk-forward replay + parallel optimizer | +| `metrics.py` | Win rate, PF, Sharpe, max DD, trade journal | + +--- + +## Disclaimer + +For educational and research use. Past performance does not guarantee +future results. Always use proper risk management. diff --git a/backtest.py b/backtest.py new file mode 100644 index 0000000..b343d47 --- /dev/null +++ b/backtest.py @@ -0,0 +1,514 @@ +""" +ForexBot2 — Walk-Forward Backtester + Parameter Optimizer +========================================================== +Replays Twelve Data USD/JPY history bar-by-bar through the same +SignalAggregator that drives live mode. The same code path that +trades USD/JPY in production is what produces these metrics — so +the backtest validates the engine, not a parallel implementation. +""" + +from __future__ import annotations + +import copy +import itertools +import logging +import multiprocessing as mp +import os +from dataclasses import replace +from typing import Dict, List, Optional, Tuple + +import pandas as pd + +from config import PlatformConfig, Timeframe, USDJPY, PIP_VALUE +from data_provider import ForexDataProvider +from metrics import BacktestResult, Trade, compute_metrics, trades_to_dataframe +from signal_aggregator import SignalAggregator, SignalReport, TradeSetup + +logger = logging.getLogger(__name__) + + +# Period in minutes for each timeframe (used for close-time computation) +TF_MINUTES: Dict[Timeframe, int] = { + Timeframe.M1: 1, + Timeframe.M5: 5, + Timeframe.M15: 15, + Timeframe.M30: 30, + Timeframe.H1: 60, + Timeframe.H4: 240, + Timeframe.D1: 1440, + Timeframe.W1: 10080, +} + +GRADE_ORDER = ["A+", "A", "B", "C", "D", "F"] + + +def _grade_passes(grade: str, min_grade: str) -> bool: + """True if `grade` is at least as good as `min_grade`.""" + if grade not in GRADE_ORDER or min_grade not in GRADE_ORDER: + return False + return GRADE_ORDER.index(grade) <= GRADE_ORDER.index(min_grade) + + +# ══════════════════════════════════════════════════════════════════ +# WALK-FORWARD BACKTESTER +# ══════════════════════════════════════════════════════════════════ + +class WalkForwardBacktester: + """Single-pair, single-position event-driven backtester.""" + + def __init__(self, config: PlatformConfig): + self.config = config + self.aggregator = SignalAggregator(config) + self.data_provider = self.aggregator.data_provider + self._regime_adx: Optional[pd.Series] = None + self._regime_tf: Optional[Timeframe] = None + + # ── Public ───────────────────────────────────────────────────── + + def load_data(self) -> Dict[Timeframe, pd.DataFrame]: + """Pull (or read from disk cache) all needed timeframes.""" + timeframes = [self.config.primary_timeframe] + list( + self.config.confirmation_timeframes + ) + bt = self.config.backtest + if bt.regime_filter and bt.regime_filter_timeframe not in timeframes: + timeframes.append(bt.regime_filter_timeframe) + + dfs: Dict[Timeframe, pd.DataFrame] = {} + for tf in timeframes: + df = self.data_provider.fetch_full_history( + symbol=self.config.symbol, + timeframe=tf, + max_bars=bt.max_bars, + start_date=bt.start_date, + end_date=bt.end_date, + ) + df = self._apply_window(df) + dfs[tf] = df + return dfs + + def run( + self, + dfs: Optional[Dict[Timeframe, pd.DataFrame]] = None, + progress: bool = True, + ) -> BacktestResult: + if dfs is None: + dfs = self.load_data() + + self._precompute_regime_filter(dfs) + + primary_tf = self.config.primary_timeframe + primary_df = dfs.get(primary_tf) + if primary_df is None or primary_df.empty: + return BacktestResult(initial_balance=self.config.backtest.initial_balance) + + # Warmup: enough bars for ema_trend & ichimoku_senkou_b on the primary TF + warmup = max( + self.config.technical.ema_trend, + self.config.technical.ichimoku_senkou_b, + 200, + ) + 30 + + if len(primary_df) <= warmup + 1: + logger.warning( + f"Insufficient data: {len(primary_df)} primary bars, " + f"need >{warmup}." + ) + return BacktestResult(initial_balance=self.config.backtest.initial_balance) + + bt = self.config.backtest + balance = bt.initial_balance + equity_points: List[Tuple[pd.Timestamp, float]] = [ + (primary_df.index[warmup], balance) + ] + trades: List[Trade] = [] + + open_position: Optional[Dict] = None + + # Optional progress bar + iterator = range(warmup, len(primary_df) - 1) + if progress: + try: + from tqdm import tqdm # type: ignore + iterator = tqdm(iterator, desc="Backtesting", unit="bar", leave=False) + except ImportError: + pass + + for t in iterator: + decision_time = self._bar_close_time(primary_df.index[t], primary_tf) + + # ── Step 1: manage any open position ─────────────────── + if open_position is not None: + next_bar = primary_df.iloc[t] + exit_info = self._check_exit(open_position, next_bar) + if exit_info is not None: + closed = self._close_trade( + open_position, next_bar.name, exit_info, balance + ) + trades.append(closed) + balance += closed.pnl_currency + equity_points.append((next_bar.name, balance)) + open_position = None + + # ── Step 2: maybe open a new position ────────────────── + if open_position is None or not bt.one_position_only: + # Slice each TF's df up to (and including) bars closed by decision_time + sliced = self._slice_dfs(dfs, decision_time) + report = self.aggregator.analyze_from_dataframes( + symbol=self.config.symbol, + dfs=sliced, + use_news=False, + ) + + if ( + _grade_passes(report.signal_quality, bt.min_trade_grade) + and abs(float(report.final_score)) >= bt.min_abs_score + and report.trade_setup is not None + and t + 1 < len(primary_df) + and self._regime_passes(decision_time) + ): + next_bar = primary_df.iloc[t + 1] + open_position = self._open_trade(report, next_bar, balance) + + # ── Force-close any still-open position at end of data ──── + if open_position is not None: + last_bar = primary_df.iloc[-1] + closed = self._close_trade( + open_position, + last_bar.name, + {"price": float(last_bar["Close"]), "reason": "end_of_data"}, + balance, + ) + trades.append(closed) + balance += closed.pnl_currency + equity_points.append((last_bar.name, balance)) + + # Build equity curve + eq_index = [p[0] for p in equity_points] + eq_values = [p[1] for p in equity_points] + equity_curve = pd.Series(eq_values, index=pd.DatetimeIndex(eq_index)) + + elapsed_days: Optional[float] = None + if len(equity_curve) >= 2: + elapsed_days = (equity_curve.index[-1] - equity_curve.index[0]).total_seconds() / 86400 + + metrics = compute_metrics( + trades, equity_curve, bt.initial_balance, elapsed_days=elapsed_days + ) + + return BacktestResult( + trades=trades, + equity_curve=equity_curve, + initial_balance=bt.initial_balance, + final_balance=balance, + metrics=metrics, + ) + + # ── Private: regime gate ─────────────────────────────────────── + + def _precompute_regime_filter(self, dfs: Dict[Timeframe, pd.DataFrame]) -> None: + """Pre-compute ADX on the regime-filter TF; cached for the whole run.""" + bt = self.config.backtest + if not bt.regime_filter: + self._regime_adx = None + self._regime_tf = None + return + + tf = bt.regime_filter_timeframe + df = dfs.get(tf) + if df is None or df.empty: + logger.warning( + f"Regime filter requested ({tf.value}) but no data available; " + f"all signals will be skipped." + ) + self._regime_adx = pd.Series(dtype=float) + self._regime_tf = tf + return + + self._regime_tf = tf + self._regime_adx = self._compute_adx_series(df, self.config.technical.adx_period) + + def _regime_passes(self, decision_time: pd.Timestamp) -> bool: + bt = self.config.backtest + if not bt.regime_filter: + return True + if self._regime_adx is None or self._regime_adx.empty: + return False + period_min = TF_MINUTES[self._regime_tf] + close_times = self._regime_adx.index + pd.Timedelta(minutes=period_min) + valid = self._regime_adx[close_times <= decision_time] + if valid.empty: + return False + last = valid.iloc[-1] + if pd.isna(last): + return False + return float(last) >= bt.regime_filter_min_adx + + @staticmethod + def _compute_adx_series(df: pd.DataFrame, period: int = 14) -> pd.Series: + """Wilder's ADX series — same algorithm as engine._compute_adx.""" + import numpy as np + high, low, close = df["High"], df["Low"], df["Close"] + tr = pd.concat([ + high - low, + (high - close.shift()).abs(), + (low - close.shift()).abs(), + ], axis=1).max(axis=1) + plus_dm = high.diff() + minus_dm = -low.diff() + plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) + minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0) + atr_s = tr.ewm(alpha=1/period, min_periods=period).mean() + plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) + minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) + dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) + return dx.ewm(alpha=1/period, min_periods=period).mean() + + # ── Private: data slicing ────────────────────────────────────── + + def _apply_window(self, df: pd.DataFrame) -> pd.DataFrame: + bt = self.config.backtest + if bt.start_date: + start = pd.Timestamp(bt.start_date, tz="UTC") + df = df[df.index >= start] + if bt.end_date: + end = pd.Timestamp(bt.end_date, tz="UTC") + df = df[df.index <= end] + return df + + @staticmethod + def _bar_close_time(open_time: pd.Timestamp, tf: Timeframe) -> pd.Timestamp: + return open_time + pd.Timedelta(minutes=TF_MINUTES[tf]) + + def _slice_dfs( + self, dfs: Dict[Timeframe, pd.DataFrame], decision_time: pd.Timestamp + ) -> Dict[Timeframe, pd.DataFrame]: + """Return only bars whose close time ≤ decision_time (no lookahead).""" + sliced: Dict[Timeframe, pd.DataFrame] = {} + for tf, df in dfs.items(): + close_times = df.index + pd.Timedelta(minutes=TF_MINUTES[tf]) + sliced[tf] = df[close_times <= decision_time] + return sliced + + # ── Private: trade execution ─────────────────────────────────── + + def _open_trade( + self, report: SignalReport, next_bar: pd.Series, balance: float + ) -> Dict: + """Open at next bar's open, applying spread + slippage.""" + ts: TradeSetup = report.trade_setup # type: ignore[assignment] + bt = self.config.backtest + + raw_entry = float(next_bar["Open"]) + + # Sanity-test inversion: flip direction and mirror SL/TP around entry + direction = ts.direction + sl = ts.stop_loss + tp = ts.take_profit_1 + if bt.invert_signal: + direction = "SELL" if direction == "BUY" else "BUY" + sl, tp = (2 * raw_entry - sl), (2 * raw_entry - tp) + + # Cost in price units = (half-spread + slippage) × pip_value + half_spread_pips = 0.5 if bt.use_spread_cost else 0.0 + cost_pips = half_spread_pips + bt.slippage_pips + cost_price = cost_pips * PIP_VALUE + + if direction == "BUY": + entry_price = raw_entry + cost_price # buy higher (worse) + else: + entry_price = raw_entry - cost_price # sell lower (worse) + + # Position sizing — risk a fixed % of current balance + risk_pct = bt.risk_per_trade_pct / 100.0 + risk_amount = balance * risk_pct + risk_pips = max(ts.risk_pips, 1e-9) + pip_currency_value = risk_amount / risk_pips # $ per pip + + return { + "direction": direction, + "entry_time": next_bar.name, + "entry_price": entry_price, + "stop_loss": sl, + "take_profit": tp, + "risk_pips": risk_pips, + "pip_currency_value": pip_currency_value, + "balance_at_open": balance, + "grade": report.signal_quality, + "score_at_entry": float(report.final_score), + "confidence_at_entry": float(report.final_confidence), + "bar_index_open": next_bar.name, + } + + def _check_exit( + self, position: Dict, bar: pd.Series + ) -> Optional[Dict]: + """Return {price, reason} if SL or TP hit, else None. + + On the entry bar (bar with same timestamp as position's entry_time), + we already filled at the open — only check subsequent bars' OHLC. + """ + if bar.name == position["bar_index_open"]: + return None # entry bar — no exit check + + bt = self.config.backtest + high = float(bar["High"]) + low = float(bar["Low"]) + + sl = position["stop_loss"] + tp = position["take_profit"] + slip = bt.slippage_pips * PIP_VALUE + half_spread = (0.5 if bt.use_spread_cost else 0.0) * PIP_VALUE + + if position["direction"] == "BUY": + sl_hit = low <= sl + tp_hit = high >= tp + if sl_hit and tp_hit: + # Conservative: SL first + return {"price": sl - slip - half_spread, "reason": "sl"} if bt.sl_first_on_tie else \ + {"price": tp - slip - half_spread, "reason": "tp"} + if sl_hit: + return {"price": sl - slip - half_spread, "reason": "sl"} + if tp_hit: + return {"price": tp - slip - half_spread, "reason": "tp"} + else: # SELL + sl_hit = high >= sl + tp_hit = low <= tp + if sl_hit and tp_hit: + return {"price": sl + slip + half_spread, "reason": "sl"} if bt.sl_first_on_tie else \ + {"price": tp + slip + half_spread, "reason": "tp"} + if sl_hit: + return {"price": sl + slip + half_spread, "reason": "sl"} + if tp_hit: + return {"price": tp + slip + half_spread, "reason": "tp"} + return None + + def _close_trade( + self, position: Dict, exit_time: pd.Timestamp, exit_info: Dict, balance: float + ) -> Trade: + exit_price = exit_info["price"] + reason = exit_info["reason"] + + if position["direction"] == "BUY": + pnl_pips = (exit_price - position["entry_price"]) / PIP_VALUE + else: + pnl_pips = (position["entry_price"] - exit_price) / PIP_VALUE + + pnl_currency = pnl_pips * position["pip_currency_value"] + pnl_pct = pnl_currency / position["balance_at_open"] * 100 + + # Duration in primary-TF bars (timestamp diff / period) + period_min = TF_MINUTES[self.config.primary_timeframe] + duration_minutes = (exit_time - position["entry_time"]).total_seconds() / 60 + duration_bars = int(round(duration_minutes / period_min)) + + return Trade( + entry_time=position["entry_time"], + exit_time=exit_time, + direction=position["direction"], + entry_price=position["entry_price"], + exit_price=exit_price, + stop_loss=position["stop_loss"], + take_profit=position["take_profit"], + exit_reason=reason, + risk_pips=position["risk_pips"], + pnl_pips=pnl_pips, + pnl_pct=pnl_pct, + pnl_currency=pnl_currency, + duration_bars=duration_bars, + grade=position["grade"], + score_at_entry=position["score_at_entry"], + confidence_at_entry=position["confidence_at_entry"], + ) + + +# ══════════════════════════════════════════════════════════════════ +# PARAMETER OPTIMIZER (PARALLEL) +# ══════════════════════════════════════════════════════════════════ + +def _apply_overrides(config: PlatformConfig, overrides: Dict[str, float]) -> PlatformConfig: + """Return a deep-copied config with parameter overrides applied.""" + new_config = copy.deepcopy(config) + + if "atr_sl_multiplier" in overrides: + new_config.signals.atr_sl_multiplier = float(overrides["atr_sl_multiplier"]) + if "default_rr_ratio" in overrides: + new_config.signals.default_rr_ratio = float(overrides["default_rr_ratio"]) + if "min_trade_grade" in overrides: + new_config.backtest.min_trade_grade = str(overrides["min_trade_grade"]) + if "min_abs_score" in overrides: + new_config.backtest.min_abs_score = float(overrides["min_abs_score"]) + if "invert_signal" in overrides: + new_config.backtest.invert_signal = bool(overrides["invert_signal"]) + if "regime_filter" in overrides: + new_config.backtest.regime_filter = bool(overrides["regime_filter"]) + if "regime_filter_min_adx" in overrides: + new_config.backtest.regime_filter_min_adx = float(overrides["regime_filter_min_adx"]) + if "risk_per_trade_pct" in overrides: + new_config.backtest.risk_per_trade_pct = float(overrides["risk_per_trade_pct"]) + + return new_config + + +def _run_single_combo(args: Tuple[PlatformConfig, Dict[str, float]]) -> BacktestResult: + """Worker function — must be top-level for multiprocessing.Pool.""" + base_config, overrides = args + config = _apply_overrides(base_config, overrides) + backtester = WalkForwardBacktester(config) + # Workers reuse the disk parquet cache, so no Twelve Data fetch needed + result = backtester.run(progress=False) + result.params = overrides + return result + + +class Optimizer: + """Embarrassingly parallel grid search over BacktestConfig.param_grid.""" + + def __init__(self, config: PlatformConfig): + self.config = config + + def run_grid( + self, + param_grid: Optional[Dict[str, List[float]]] = None, + prewarm_cache: bool = True, + ) -> List[BacktestResult]: + if param_grid is None: + param_grid = self.config.backtest.param_grid + + # Prewarm the disk cache once before fanning out — avoids each worker + # re-fetching from Twelve Data. + if prewarm_cache: + logger.info("Prewarming disk cache for backtest data...") + warmer = WalkForwardBacktester(self.config) + warmer.load_data() + + keys = list(param_grid.keys()) + value_lists = [param_grid[k] for k in keys] + combos = [ + dict(zip(keys, vals)) + for vals in itertools.product(*value_lists) + ] + + n_workers = min( + self.config.backtest.parallel_workers, + len(combos), + os.cpu_count() or 1, + ) + logger.info( + f"Optimizer: {len(combos)} combos × {n_workers} workers" + ) + + tasks = [(self.config, c) for c in combos] + + if n_workers <= 1: + results = [_run_single_combo(t) for t in tasks] + else: + with mp.Pool(processes=n_workers) as pool: + results = pool.map(_run_single_combo, tasks) + + # Sort by total return descending + results.sort( + key=lambda r: r.metrics.get("total_return_pct", -float("inf")), + reverse=True, + ) + return results diff --git a/config.py b/config.py index 8bb648b..65cd094 100644 --- a/config.py +++ b/config.py @@ -1,13 +1,15 @@ """ -Forex Signal Intelligence Platform - Configuration -=================================================== -Central configuration for all signal analysis parameters. -Tuned by market regime and validated against historical performance. +ForexBot2 — Configuration (USD/JPY focus) +========================================== +Single-pair signal & backtest configuration. +All knobs that govern indicator behavior, signal scoring, news handling, +and the walk-forward backtester live here. """ +import os from dataclasses import dataclass, field from enum import Enum -from typing import Dict, List +from typing import Dict, List, Optional class Timeframe(Enum): @@ -37,160 +39,160 @@ class MarketSession(Enum): OVERLAP_LONDON_NY = "london_ny_overlap" -# ── Major Forex Pairs ────────────────────────────────────────────── -MAJOR_PAIRS = [ - "EURUSD=X", "GBPUSD=X", "USDJPY=X", "USDCHF=X", - "AUDUSD=X", "USDCAD=X", "NZDUSD=X", -] - -CROSS_PAIRS = [ - "EURGBP=X", "EURJPY=X", "GBPJPY=X", "AUDJPY=X", - "EURCHF=X", "CADJPY=X", -] - -# Yahoo Finance symbol mapping for display -SYMBOL_DISPLAY = { - "EURUSD=X": "EUR/USD", "GBPUSD=X": "GBP/USD", - "USDJPY=X": "USD/JPY", "USDCHF=X": "USD/CHF", - "AUDUSD=X": "AUD/USD", "USDCAD=X": "USD/CAD", - "NZDUSD=X": "NZD/USD", "EURGBP=X": "EUR/GBP", - "EURJPY=X": "EUR/JPY", "GBPJPY=X": "GBP/JPY", - "AUDJPY=X": "AUD/JPY", "EURCHF=X": "EUR/CHF", - "CADJPY=X": "CAD/JPY", -} +# ── Single Focus: USD/JPY ───────────────────────────────────────── +USDJPY = "USDJPY=X" +DISPLAY = "USD/JPY" +PIP_VALUE = 0.01 # JPY pairs use 0.01 per pip @dataclass class TechnicalConfig: - """Technical indicator parameters - carefully tuned for Forex.""" - # EMA periods (fast/medium/slow for trend detection) + """Technical indicator parameters.""" ema_fast: int = 9 ema_medium: int = 21 ema_slow: int = 50 ema_trend: int = 200 - # RSI rsi_period: int = 14 rsi_overbought: float = 70.0 rsi_oversold: float = 30.0 rsi_neutral_upper: float = 60.0 rsi_neutral_lower: float = 40.0 - # MACD (12, 26, 9 is standard) macd_fast: int = 12 macd_slow: int = 26 macd_signal: int = 9 - # Bollinger Bands bb_period: int = 20 bb_std: float = 2.0 - # ATR for volatility measurement atr_period: int = 14 - # Stochastic stoch_k: int = 14 stoch_d: int = 3 stoch_overbought: float = 80.0 stoch_oversold: float = 20.0 - # ADX for trend strength adx_period: int = 14 adx_strong_trend: float = 25.0 adx_very_strong: float = 50.0 - # Ichimoku ichimoku_tenkan: int = 9 ichimoku_kijun: int = 26 ichimoku_senkou_b: int = 52 - # Volume profile volume_ma_period: int = 20 - - # Pivot points lookback pivot_lookback: int = 1 - - # Divergence detection — how many bars back to scan for swing points divergence_lookback: int = 14 @dataclass class SignalConfig: - """Signal scoring and filtering parameters.""" - # Minimum confidence to emit a signal (0-100) + """Signal scoring & filtering.""" min_confidence: float = 55.0 - # Weight allocation for composite score weights: Dict[str, float] = field(default_factory=lambda: { - "trend": 0.25, # EMA alignment, ADX - "momentum": 0.20, # RSI, Stochastic, MACD - "volatility": 0.15, # BB position, ATR regime - "price_action": 0.15, # Candlestick patterns, S/R - "multi_timeframe": 0.15,# Higher TF confirmation - "sentiment": 0.10, # News/calendar events + "trend": 0.25, + "momentum": 0.20, + "volatility": 0.15, + "price_action": 0.15, + "multi_timeframe": 0.15, + "sentiment": 0.10, }) - # Multi-timeframe confirmation requirements - mtf_agreement_threshold: int = 2 # out of 3 timeframes must agree - - # Signal cooldown (prevent over-signalling same direction) + mtf_agreement_threshold: int = 2 # of 3 timeframes cooldown_bars: int = 5 - # Risk management defaults - default_risk_pct: float = 1.0 # risk 1% per trade - default_rr_ratio: float = 2.0 # 1:2 risk-reward minimum + default_risk_pct: float = 1.0 + default_rr_ratio: float = 2.0 + atr_sl_multiplier: float = 1.5 # SL distance = 1.5 × ATR @dataclass class NewsConfig: - """Economic calendar and sentiment configuration.""" - # High-impact event currencies - tracked_currencies: List[str] = field(default_factory=lambda: [ - "USD", "EUR", "GBP", "JPY", "CHF", "AUD", "CAD", "NZD" - ]) - - # Event impact levels to monitor - min_impact: str = "medium" # "low", "medium", "high" - - # Pre-event buffer (minutes before high-impact event = no new signals) + """Economic calendar & sentiment.""" + tracked_currencies: List[str] = field(default_factory=lambda: ["USD", "JPY"]) + min_impact: str = "medium" pre_event_buffer_minutes: int = 30 - - # Post-event volatility window (minutes after event = cautious signals) post_event_buffer_minutes: int = 15 - # News sentiment sources rss_feeds: List[str] = field(default_factory=lambda: [ "https://www.forexlive.com/feed/", "https://www.fxstreet.com/rss", ]) +@dataclass +class BacktestConfig: + """Walk-forward backtest configuration.""" + # Data window — None means "all available history" + start_date: Optional[str] = None # "YYYY-MM-DD" + end_date: Optional[str] = None # "YYYY-MM-DD" + max_bars: int = 5000 # Twelve Data ceiling per request + + # Account & sizing + initial_balance: float = 10_000.0 + risk_per_trade_pct: float = 1.0 # % of equity risked per trade + + # Execution realism + use_spread_cost: bool = True + slippage_pips: float = 0.5 # added to spread on entry & exit + one_position_only: bool = True + sl_first_on_tie: bool = True # if SL & TP both inside bar, SL hits first + + # Signal gates + min_trade_grade: str = "B" # "A+", "A", "B", "C", "D", "F" + min_abs_score: float = 0.0 # |final_score| must exceed this to enter + + # Sanity-test knob — flip BUY↔SELL to see if the engine has reversed edge + invert_signal: bool = False + + # Regime gate — only trade when a higher-TF trend is confirmed + regime_filter: bool = False + regime_filter_timeframe: Timeframe = Timeframe.H4 + regime_filter_min_adx: float = 25.0 + + # Parallelism (used by Optimizer; single-run backtest is sequential) + parallel_workers: int = field(default_factory=lambda: os.cpu_count() or 4) + + # Optimizer + optimize: bool = False + param_grid: Dict[str, list] = field(default_factory=lambda: { + # 2 × 2 × 3 × 2 × 2 = 48 combos + "atr_sl_multiplier": [1.5, 2.5], + "default_rr_ratio": [1.5, 2.5], + "min_abs_score": [15.0, 35.0, 60.0], + "min_trade_grade": ["B", "A"], + "invert_signal": [False, True], + }) + + # Cache directory for fetched OHLCV (parquet) + cache_dir: str = ".cache" + + @dataclass class PlatformConfig: - """Overall platform configuration.""" + """Top-level configuration.""" technical: TechnicalConfig = field(default_factory=TechnicalConfig) signals: SignalConfig = field(default_factory=SignalConfig) news: NewsConfig = field(default_factory=NewsConfig) + backtest: BacktestConfig = field(default_factory=BacktestConfig) - # Default pairs to monitor - watch_pairs: List[str] = field(default_factory=lambda: MAJOR_PAIRS) + symbol: str = USDJPY + display_name: str = DISPLAY + pip_value: float = PIP_VALUE - # Analysis timeframes (primary + confirmation) primary_timeframe: Timeframe = Timeframe.H1 confirmation_timeframes: List[Timeframe] = field( default_factory=lambda: [Timeframe.M15, Timeframe.H4] ) - # Data source data_lookback_days: int = 60 # enough for 200 EMA on H1 - # Twelve Data API key (overridden by TWELVE_DATA_API_KEY env var) twelve_data_api_key: str = "f563b22ffdfd4924955decbe1a8c8701" - # Dashboard refresh interval (seconds) - refresh_interval: int = 300 # 5 minutes + refresh_interval: int = 300 # 5 min cache TTL - # Logging log_level: str = "INFO" log_signals_to_file: bool = True signal_log_path: str = "signals.log" diff --git a/data_provider.py b/data_provider.py index ab0b4ab..07cfd40 100644 --- a/data_provider.py +++ b/data_provider.py @@ -1,23 +1,22 @@ """ -Forex Signal Intelligence Platform - Data Provider -==================================================== -Retrieves and validates market data from Twelve Data API. -Falls back to synthetic GBM data if the API is unavailable. +ForexBot2 — Data Provider (Twelve Data, USD/JPY only) +======================================================= +Fetches and validates OHLCV from Twelve Data. No synthetic fallback — +if the API fails the caller learns about it. Adds a parquet on-disk +cache so the backtest worker pool reuses one download instead of +hammering Twelve Data 27 times during a parameter sweep. """ -import hashlib import logging import os import time from collections import deque -from datetime import datetime, timedelta, timezone -from typing import Dict, List, Optional +from typing import Dict, Optional -import numpy as np import pandas as pd import requests -from config import PlatformConfig, Timeframe, SYMBOL_DISPLAY +from config import PlatformConfig, Timeframe, USDJPY, DISPLAY, PIP_VALUE logger = logging.getLogger(__name__) @@ -28,12 +27,7 @@ class DataIntegrityError(Exception): class ForexDataProvider: - """ - 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. - """ + """Twelve Data REST client + on-disk parquet cache.""" TD_BASE_URL = "https://api.twelvedata.com" @@ -48,7 +42,6 @@ class ForexDataProvider: Timeframe.W1: "1week", } - # Approximate bars per day for each timeframe (used to size outputsize) BARS_PER_DAY = { Timeframe.M1: 1440, Timeframe.M5: 288, @@ -60,7 +53,7 @@ class ForexDataProvider: Timeframe.W1: 0.143, } - # Free plan: 8 API credits per minute (each symbol = 1 credit) + # Free plan: 8 credits / minute RATE_LIMIT = 8 RATE_WINDOW = 60.0 @@ -72,7 +65,7 @@ class ForexDataProvider: ) self._cache: Dict[str, pd.DataFrame] = {} self._cache_timestamps: Dict[str, float] = {} - self._request_times: deque = deque() # timestamps of recent API credits used + self._request_times: deque = deque() self._session = requests.Session() self._session.headers.update({ "User-Agent": "Mozilla/5.0 (compatible; ForexIntel/1.0)" @@ -80,22 +73,18 @@ class ForexDataProvider: @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:]}" + # ── Public API ───────────────────────────────────────────────── + def fetch_ohlcv( self, - symbol: str, - timeframe: Timeframe, + symbol: str = USDJPY, + timeframe: Timeframe = Timeframe.H1, lookback_days: Optional[int] = None, ) -> pd.DataFrame: - """ - Fetch OHLCV data for a Forex pair with full validation. - - Returns a DataFrame with columns: Open, High, Low, Close, Volume - Index is a timezone-aware DatetimeIndex in UTC. - """ + """Fetch OHLCV with in-memory caching for live mode.""" if lookback_days is None: lookback_days = self.config.data_lookback_days @@ -111,55 +100,63 @@ class ForexDataProvider: if cached is not None and not cached.empty: return cached - df = self._fetch_twelve_data(symbol, timeframe, lookback_days) + outputsize = max( + 300, + min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20, 5000), + ) + df = self._fetch_twelve_data(symbol, timeframe, outputsize) df = self._validate_and_clean(df, symbol, timeframe) self._cache[cache_key] = df self._cache_timestamps[cache_key] = now return df - def prefetch_batch( + def fetch_full_history( self, - symbols: List[str], - timeframes: List[Timeframe], - lookback_days: Optional[int] = None, - ) -> None: + symbol: str = USDJPY, + timeframe: Timeframe = Timeframe.H1, + max_bars: int = 5000, + use_disk_cache: bool = True, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + ) -> pd.DataFrame: """ - 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. + Pull the largest window Twelve Data allows in one call. + On-disk parquet cache prevents re-download across worker subprocesses + during an optimizer sweep. """ - if lookback_days is None: - lookback_days = self.config.data_lookback_days + cache_path = self._disk_cache_path(symbol, timeframe, start_date, end_date) - for tf in timeframes: + if use_disk_cache and os.path.exists(cache_path): try: - self._batch_fetch_timeframe(symbols, tf, lookback_days) - time.sleep(1.0) # brief pause between TF requests + df = pd.read_parquet(cache_path) + if df.index.tz is None: + df.index = df.index.tz_localize("UTC") + logger.info( + f"Loaded {len(df)} bars from disk cache: {cache_path}" + ) + return df except Exception as e: - logger.warning(f"Batch prefetch failed for {tf.value}: {e}") + logger.warning(f"Disk cache read failed ({e}); refetching.") - def fetch_multiple( - self, - symbols: List[str], - timeframe: Timeframe, - lookback_days: Optional[int] = None, - ) -> Dict[str, pd.DataFrame]: - """Fetch data for multiple symbols, returning dict of DataFrames.""" - results = {} - for symbol in symbols: + outputsize = min(max_bars, 5000) + df = self._fetch_twelve_data( + symbol, timeframe, outputsize, + start_date=start_date, end_date=end_date, + ) + df = self._validate_and_clean(df, symbol, timeframe) + + if use_disk_cache: + os.makedirs(os.path.dirname(cache_path), exist_ok=True) try: - df = self.fetch_ohlcv(symbol, timeframe, lookback_days) - results[symbol] = df + df.to_parquet(cache_path) + logger.info(f"Cached {len(df)} bars → {cache_path}") except Exception as e: - logger.warning(f"Failed to fetch {symbol}: {e}") - return results + logger.warning(f"Disk cache write failed: {e}") - def get_latest_price(self, symbol: str) -> Optional[float]: - """Return the most recent close price for a symbol.""" + return df + + def get_latest_price(self, symbol: str = USDJPY) -> Optional[float]: try: df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) if df is not None and not df.empty: @@ -168,47 +165,47 @@ class ForexDataProvider: pass return None - def get_spread_estimate(self, symbol: str) -> float: - """Estimate typical spread from recent M5 high-low ranges.""" + def get_spread_estimate(self, symbol: str = USDJPY) -> float: + """Estimate spread from recent M5 H-L median; fallback to 1 pip on JPY.""" try: df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) if df is not None and not df.empty: return float((df["High"] - df["Low"]).median()) except Exception: pass + # Fallback: 1 pip = 0.01 for JPY pairs + return PIP_VALUE - fallback_spreads = { - "EURUSD=X": 0.00010, "GBPUSD=X": 0.00015, - "USDJPY=X": 0.010, "USDCHF=X": 0.00015, - "AUDUSD=X": 0.00015, "USDCAD=X": 0.00018, - "NZDUSD=X": 0.00020, - } - 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 + def get_pair_info(self, symbol: str = USDJPY) -> Dict: return { "symbol": symbol, - "display": display, - "pip_value": 0.01 if is_jpy else 0.0001, - "is_jpy_pair": is_jpy, + "display": DISPLAY, + "pip_value": PIP_VALUE, + "is_jpy_pair": True, } - # ── Private Methods ──────────────────────────────────────────── + # ── Private ──────────────────────────────────────────────────── + + def _disk_cache_path( + self, + symbol: str, + timeframe: Timeframe, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + ) -> str: + clean = symbol.replace("=X", "").lower() + suffix = "" + if start_date or end_date: + suffix = f"_{start_date or 'min'}_{end_date or 'max'}" + return os.path.join( + self.config.backtest.cache_dir, + f"{clean}_{timeframe.value}{suffix}.parquet", + ) 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. - """ + """Free tier: 8 credits / minute. Block until budget allows.""" 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() @@ -218,276 +215,82 @@ class ForexDataProvider: 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..." + f"— waiting {wait:.1f}s..." ) 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. - """ + def _fetch_twelve_data( + self, + symbol: str, + timeframe: Timeframe, + outputsize: int, + start_date: Optional[str] = None, + end_date: Optional[str] = None, + ) -> pd.DataFrame: if not self._api_key: - return + raise DataIntegrityError( + "No Twelve Data API key. Set TWELVE_DATA_API_KEY env var " + "or config.twelve_data_api_key." + ) - td_symbols = [self._to_td_symbol(s) for s in symbols] + 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) - # Each symbol in a batch request costs 1 credit - self._throttle(credits=len(symbols)) + self._throttle(credits=1) + + params = { + "symbol": td_symbol, + "interval": interval, + "outputsize": outputsize, + "apikey": self._api_key, + "timezone": "UTC", + "format": "json", + } + if start_date: + params["start_date"] = start_date + if end_date: + params["end_date"] = end_date 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", - }, + params=params, 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": + if data.get("status") == "error": raise DataIntegrityError( - f"Batch API error ({timeframe.value}): {data.get('message', 'unknown')}" + f"Twelve Data error for {td_symbol}: {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 = data.get("values") + if not values: + raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") - 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) + ] - 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 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: - 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 data.get("status") == "error": - raise DataIntegrityError( - f"Twelve Data API error for {td_symbol}: {data.get('message', 'unknown')}" - ) - - values = data.get("values") - if not values: - raise DataIntegrityError(f"Empty response from Twelve Data for {td_symbol}") - - # 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) - ] - - df = pd.DataFrame(records).set_index("Datetime") - df.index.name = "Datetime" - logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") - return df - - except DataIntegrityError: - raise - except Exception as e: - 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 using GBM with mean-reversion. - Used as a fallback when live data is unavailable. - """ - 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}, - "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}, - "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}, - } - - params = pair_params.get(symbol, {"price": 1.0000, "vol": 0.08, "drift": 0.000}) - base_price = params["price"] - annual_vol = params["vol"] - annual_drift = params["drift"] - - tf_minutes = { - Timeframe.M1: 1, Timeframe.M5: 5, Timeframe.M15: 15, - Timeframe.M30: 30, Timeframe.H1: 60, Timeframe.H4: 240, - Timeframe.D1: 1440, Timeframe.W1: 10080, - } - bar_minutes = tf_minutes.get(timeframe, 60) - bars_per_year = (365.25 * 24 * 60) / bar_minutes - dt = 1.0 / bars_per_year - - n_bars = min(int(lookback_days * 24 * 60 / bar_minutes), 2000) - n_bars = max(n_bars, 300) - - seed_base = int(hashlib.md5(symbol.encode()).hexdigest()[:8], 16) - time_component = int(time.time()) // 3600 - np.random.seed((seed_base + time_component) % (2**31)) - - bar_vol = annual_vol * np.sqrt(dt) - bar_drift = annual_drift * dt - - regime_len = np.random.randint(20, 80) - trend_strength = np.random.normal(0, 0.3) - - price = base_price - prices = [price] - - for i in range(1, n_bars): - if i % regime_len == 0: - regime_len = np.random.randint(20, 80) - trend_strength = np.random.normal(0, 0.3) - - mr_force = -0.001 * (price - base_price) / base_price - ret = bar_drift + trend_strength * dt + mr_force + bar_vol * np.random.randn() - - if np.random.random() < 0.02: - ret *= np.random.uniform(2.0, 4.0) - - price *= (1 + ret) - prices.append(price) - - prices = np.array(prices) - opens = np.roll(prices, 1) - opens[0] = prices[0] - - 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)) - highs = np.maximum(highs, np.maximum(opens, prices)) - lows = np.minimum(lows, np.minimum(opens, prices)) - - volume = np.random.lognormal(mean=10, sigma=0.5, size=n_bars).astype(int) - - freq_map = { - Timeframe.M1: "1min", Timeframe.M5: "5min", - Timeframe.M15: "15min", Timeframe.M30: "30min", - Timeframe.H1: "1h", Timeframe.H4: "4h", - Timeframe.D1: "D", Timeframe.W1: "W", - } - freq = freq_map.get(timeframe, "1h") - 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, - }, index=index) + df = pd.DataFrame(records).set_index("Datetime") df.index.name = "Datetime" - - logger.info( - f"Generated {n_bars} synthetic {timeframe.value} bars for " - f"{SYMBOL_DISPLAY.get(symbol, symbol)} @ {prices[-1]:.5f}" - ) + logger.debug(f"Fetched {len(df)} bars for {td_symbol} @ {interval}") return df def _validate_and_clean( self, df: pd.DataFrame, symbol: str, timeframe: Timeframe ) -> pd.DataFrame: - """ - 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}") diff --git a/main.py b/main.py index d5a0c5f..2e0e184 100644 --- a/main.py +++ b/main.py @@ -1,509 +1,457 @@ -#!/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 +#!/usr/bin/env python3 +""" +ForexBot2 — USD/JPY Signal Intelligence (single-pair) +======================================================= +Entry point for both live signal analysis and walk-forward backtesting. + +Usage: + python main.py # Live signal on H1 + python main.py --timeframe H4 --detail # Detailed H4 signal + python main.py --json # JSON output + python main.py --backtest # Walk-forward backtest + python main.py --backtest --optimize # Parallel grid search +""" + +import argparse +import json +import logging +import sys +from datetime import datetime, timezone +from typing import Dict, List + +# Force UTF-8 output on Windows so box-drawing chars render +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, USDJPY, DISPLAY +from signal_aggregator import SignalAggregator, SignalReport + + +# ══════════════════════════════════════════════════════════════════ +# TERMINAL FORMATTING +# ══════════════════════════════════════════════════════════════════ + +class Colors: + 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" + + +def signal_color(signal: SignalStrength) -> str: + 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(q: str) -> str: + return { + "A+": Colors.GREEN + Colors.BOLD, + "A": Colors.GREEN, + "B": Colors.CYAN, + "C": Colors.YELLOW, + "D": Colors.GRAY, + "F": Colors.DIM, + }.get(q, Colors.RESET) + + +def print_header(title: str = "USD/JPY SIGNAL INTELLIGENCE"): + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} {title}{Colors.RESET}") + print(f"{Colors.GRAY} Time: {now}{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") + + +# ── Live signal renderer ─────────────────────────────────────────── + +def print_signal_report(report: SignalReport, detail: bool = True): + sc = signal_color(report.final_signal) + qc = quality_color(report.signal_quality) + sig_name = report.final_signal.name.replace("_", " ") + + print(f" {Colors.BOLD}{Colors.WHITE}{report.display_name}{Colors.RESET}" + f" │ Price: {Colors.BOLD}{report.current_price:.3f}{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}") + + if report.mtf_signals: + mtf_status = (f" {Colors.GREEN}✓ ALIGNED{Colors.RESET}" + if report.mtf_agreement + else f" {Colors.YELLOW}✗ DIVERGENT{Colors.RESET}") + print(f"\n {Colors.BOLD}Multi-Timeframe:{Colors.RESET}{mtf_status}") + 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}" + f" ({tf_score:+.1f})") + + if detail and 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) + print(f" {ind.name:<18} {isc}{ind.signal.name.replace('_',' '):<12}{Colors.RESET}" + f" ({ind.confidence:.0f}%) │ {Colors.GRAY}{ind.description}{Colors.RESET}") + + if report.bullish_reasons: + print(f"\n {Colors.GREEN}▲ Bullish Factors:{Colors.RESET}") + for r in report.bullish_reasons[:5]: + print(f" {Colors.GREEN}+{Colors.RESET} {r}") + if report.bearish_reasons: + print(f"\n {Colors.RED}▼ Bearish Factors:{Colors.RESET}") + for r in report.bearish_reasons[:5]: + print(f" {Colors.RED}-{Colors.RESET} {r}") + + if report.warnings: + print(f"\n {Colors.YELLOW}⚠ Warnings:{Colors.RESET}") + for w in report.warnings: + print(f" {Colors.YELLOW}!{Colors.RESET} {w}") + + 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}") + + 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:.3f}") + print(f" Stop Loss: {Colors.RED}{ts.stop_loss:.3f}{Colors.RESET} ({ts.risk_pips:.1f} pips)") + print(f" Take Profit 1: {Colors.GREEN}{ts.take_profit_1:.3f}{Colors.RESET} ({ts.reward_pips:.1f} pips)") + print(f" Take Profit 2: {Colors.GREEN}{ts.take_profit_2:.3f}{Colors.RESET}") + print(f" Risk:Reward: 1:{ts.risk_reward:.1f}") + + 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}") + for lvl in ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"]: + if lvl in levels: + val = levels[lvl] + color = Colors.RED if val > report.current_price else Colors.GREEN + print(f" {lvl.upper():>6}: {color}{val:.3f}{Colors.RESET}") + + +def report_to_dict(report: SignalReport) -> 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"] = { + name: { + "signal": ind.signal.name, + "confidence": round(ind.confidence, 1), + "description": ind.description, + } + for name, ind in report.primary_analysis.indicators.items() + if not name.startswith("_") + } + d["key_levels"] = { + k: round(v, 5) for k, v in report.primary_analysis.key_levels.items() + } + return d + + +# ── Backtest renderer ────────────────────────────────────────────── + +def print_backtest_report(result, params: dict = None, verbose_journal: bool = False): + m = result.metrics + print(f" {Colors.BOLD}{Colors.WHITE}Backtest Summary — USD/JPY{Colors.RESET}") + if params: + params_str = ", ".join(f"{k}={v}" for k, v in params.items()) + print(f" {Colors.GRAY}Params: {params_str}{Colors.RESET}") + print(f" {Colors.GRAY}{'─' * 76}{Colors.RESET}") + + initial = result.initial_balance + final = result.final_balance + ret_pct = m.get("total_return_pct", 0) + ret_color = Colors.GREEN if ret_pct >= 0 else Colors.RED + + print(f" Trades: {Colors.BOLD}{m.get('total_trades', 0)}{Colors.RESET}" + f" ({m.get('wins', 0)}W / {m.get('losses', 0)}L)") + print(f" Win Rate: {m.get('win_rate', 0):.1f}%") + print(f" Profit Factor: {m.get('profit_factor', 0):.2f}") + pf_inf = m.get('profit_factor', 0) == float('inf') + if pf_inf: + print(f" (no losing trades — infinite PF)") + print(f" Expectancy: {m.get('expectancy_pips', 0):+.2f} pips/trade" + f" ({m.get('expectancy_per_trade', 0):+.2f} per trade)") + print(f" Initial Balance: ${initial:,.2f}") + print(f" Final Balance: {ret_color}${final:,.2f}{Colors.RESET}") + print(f" Total Return: {ret_color}{ret_pct:+.2f}%{Colors.RESET}") + print(f" CAGR: {m.get('cagr', 0):+.2f}%") + print(f" Max Drawdown: {Colors.RED}{m.get('max_drawdown_pct', 0):.2f}%{Colors.RESET}") + print(f" Sharpe Ratio: {m.get('sharpe', 0):.2f}") + print(f" Sortino Ratio: {m.get('sortino', 0):.2f}") + print(f" Avg Duration: {m.get('avg_trade_duration_bars', 0):.1f} bars") + print(f" Win Streak: {m.get('longest_winning_streak', 0)} | " + f"Loss Streak: {m.get('longest_losing_streak', 0)}") + print(f" {Colors.GRAY}{'─' * 76}{Colors.RESET}") + + if verbose_journal and result.trades: + print(f"\n {Colors.BOLD}Trade Journal (last 10):{Colors.RESET}") + print(f" {'#':<4} {'Entry Time':<22} {'Dir':<5} {'Entry':<9} " + f"{'Exit':<9} {'Reason':<6} {'Pips':>8} {'P&L%':>7}") + for i, t in enumerate(result.trades[-10:], start=max(1, len(result.trades) - 9)): + color = Colors.GREEN if t.pnl_currency > 0 else Colors.RED + entry_str = str(t.entry_time)[:19] + print(f" {i:<4} {entry_str:<22} {t.direction:<5} {t.entry_price:<9.3f} " + f"{t.exit_price:<9.3f} {t.exit_reason:<6} " + f"{color}{t.pnl_pips:>+8.1f}{Colors.RESET} " + f"{color}{t.pnl_pct:>+6.2f}%{Colors.RESET}") + + +# ══════════════════════════════════════════════════════════════════ +# CLI +# ══════════════════════════════════════════════════════════════════ + +def parse_args(): + parser = argparse.ArgumentParser( + description="ForexBot2 — USD/JPY Signal Intelligence + Backtester", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + python main.py Live signal (H1, summary) + python main.py --detail Live signal with full breakdown + python main.py --timeframe H4 --detail H4 timeframe + python main.py --json JSON output + python main.py --backtest Walk-forward backtest (full history) + python main.py --backtest --bars 2000 Backtest on last 2000 bars + python main.py --backtest --optimize Parallel grid search across CPU cores + """, + ) + parser.add_argument("--timeframe", "-tf", default="H1", + choices=["M5", "M15", "M30", "H1", "H4", "D1"], + help="Primary analysis timeframe (default: H1)") + parser.add_argument("--detail", action="store_true", + help="Show full indicator breakdown (live mode)") + parser.add_argument("--json", action="store_true", + help="Output as JSON") + parser.add_argument("--verbose", "-v", action="store_true", + help="Debug logging") + + # Backtest options + parser.add_argument("--backtest", action="store_true", + help="Run walk-forward backtest instead of live signal") + parser.add_argument("--optimize", action="store_true", + help="Run parallel parameter grid search (implies --backtest)") + parser.add_argument("--bars", type=int, default=5000, + help="Max bars of history (default 5000 — TD ceiling)") + parser.add_argument("--start", default=None, + help="Backtest window start (YYYY-MM-DD)") + parser.add_argument("--end", default=None, + help="Backtest window end (YYYY-MM-DD)") + parser.add_argument("--min-grade", default="B", + choices=["A+", "A", "B", "C", "D", "F"], + help="Minimum grade to open backtest trade (default B)") + parser.add_argument("--journal", action="store_true", + help="Print last 10 trades from backtest journal") + parser.add_argument("--journal-csv", default=None, + help="Write full trade journal to CSV path") + parser.add_argument("--regime-filter", action="store_true", + help="Gate signals on higher-TF ADX trend confirmation") + parser.add_argument("--regime-min-adx", type=float, default=30.0, + help="Min ADX for regime filter (default 30, requires --regime-filter)") + + return parser.parse_args() + + +def build_config(args) -> PlatformConfig: + config = PlatformConfig() + + 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] + + 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] + ) + + # Backtest config + config.backtest.start_date = args.start + config.backtest.end_date = args.end + config.backtest.max_bars = args.bars + config.backtest.min_trade_grade = args.min_grade + config.backtest.optimize = args.optimize + config.backtest.regime_filter = args.regime_filter + config.backtest.regime_filter_min_adx = args.regime_min_adx + + return config + + +def run_live(args, config: PlatformConfig): + print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") + aggregator = SignalAggregator(config) + report = aggregator.analyze_pair() + + if args.json: + print(json.dumps(report_to_dict(report), indent=2)) + return + + print_header("USD/JPY SIGNAL INTELLIGENCE") + print_signal_report(report, detail=args.detail) + print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") + print(f"{Colors.GRAY} Disclaimer: For educational/research purposes only.{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") + + +def run_backtest(args, config: PlatformConfig): + from backtest import WalkForwardBacktester + from metrics import trades_to_dataframe + + print_header("USD/JPY WALK-FORWARD BACKTEST") + print(f"{Colors.GRAY} Loading data and replaying bars...{Colors.RESET}\n") + + backtester = WalkForwardBacktester(config) + result = backtester.run() + + if args.json: + out = { + "metrics": result.metrics, + "initial_balance": result.initial_balance, + "final_balance": result.final_balance, + "trades": [ + { + "entry_time": str(t.entry_time), + "exit_time": str(t.exit_time), + "direction": t.direction, + "entry_price": t.entry_price, + "exit_price": t.exit_price, + "exit_reason": t.exit_reason, + "pnl_pips": t.pnl_pips, + "pnl_pct": t.pnl_pct, + "grade": t.grade, + } + for t in result.trades + ], + } + print(json.dumps(out, indent=2)) + else: + print_backtest_report(result, verbose_journal=args.journal) + + if args.journal_csv and result.trades: + df = trades_to_dataframe(result.trades) + df.to_csv(args.journal_csv, index=False) + print(f"\n {Colors.GREEN}✓ Trade journal written to {args.journal_csv}{Colors.RESET}") + + +def run_optimize(args, config: PlatformConfig): + from backtest import Optimizer + + print_header("USD/JPY BACKTEST PARAMETER OPTIMIZER") + grid = config.backtest.param_grid + n_combos = 1 + for v in grid.values(): + n_combos *= len(v) + print(f"{Colors.GRAY} Grid: {n_combos} combos × {config.backtest.parallel_workers} workers{Colors.RESET}\n") + + optimizer = Optimizer(config) + results = optimizer.run_grid(grid) + + print(f"\n {Colors.BOLD}Top 10 Parameter Combinations (by total return):{Colors.RESET}\n") + print(f" {'#':<3} {'Return%':>8} {'PF':>6} {'WinR':>6} {'Trades':>7} " + f"{'MaxDD%':>7} {'Sharpe':>7} Params") + print(f" {Colors.GRAY}{'─' * 90}{Colors.RESET}") + + for i, r in enumerate(results[:10], start=1): + m = r.metrics + ret = m.get("total_return_pct", 0) + ret_color = Colors.GREEN if ret >= 0 else Colors.RED + params_str = ", ".join(f"{k}={v}" for k, v in r.params.items()) + pf = m.get("profit_factor", 0) + pf_str = "inf" if pf == float("inf") else f"{pf:.2f}" + print(f" {i:<3} {ret_color}{ret:>+7.2f}%{Colors.RESET} {pf_str:>6} " + f"{m.get('win_rate', 0):>5.1f}% {m.get('total_trades', 0):>7} " + f"{m.get('max_drawdown_pct', 0):>+6.2f}% {m.get('sharpe', 0):>7.2f}" + f" {Colors.GRAY}{params_str}{Colors.RESET}") + + print(f"\n {Colors.BOLD}Best combination details:{Colors.RESET}\n") + print_backtest_report(results[0], params=results[0].params, verbose_journal=False) + + +def main(): + args = parse_args() + + 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", + ) + + config = build_config(args) + + if args.optimize: + run_optimize(args, config) + elif args.backtest: + run_backtest(args, config) + else: + run_live(args, config) + + +if __name__ == "__main__": + main() diff --git a/metrics.py b/metrics.py new file mode 100644 index 0000000..261b89e --- /dev/null +++ b/metrics.py @@ -0,0 +1,155 @@ +""" +ForexBot2 — Backtest Metrics +============================== +Pure functions over Trade records & equity curves. +""" + +from dataclasses import dataclass, field, asdict +from typing import Dict, List, Optional + +import numpy as np +import pandas as pd + + +@dataclass +class Trade: + """A single closed trade.""" + entry_time: pd.Timestamp + exit_time: pd.Timestamp + direction: str # "BUY" / "SELL" + entry_price: float + exit_price: float + stop_loss: float + take_profit: float + exit_reason: str # "tp" / "sl" / "end_of_data" + risk_pips: float + pnl_pips: float + pnl_pct: float # P&L as % of equity at trade open + pnl_currency: float # P&L in account currency (units = balance) + duration_bars: int + grade: str # signal grade at entry + score_at_entry: float + confidence_at_entry: float + + +@dataclass +class BacktestResult: + """Complete backtest output.""" + trades: List[Trade] = field(default_factory=list) + equity_curve: pd.Series = field(default_factory=lambda: pd.Series(dtype=float)) + initial_balance: float = 0.0 + final_balance: float = 0.0 + metrics: Dict[str, float] = field(default_factory=dict) + params: Dict[str, float] = field(default_factory=dict) + + +def compute_metrics( + trades: List[Trade], + equity_curve: pd.Series, + initial_balance: float, + elapsed_days: Optional[float] = None, +) -> Dict[str, float]: + """All summary metrics from a list of closed trades + equity curve.""" + if not trades: + return { + "total_trades": 0, + "win_rate": 0.0, + "profit_factor": 0.0, + "expectancy_per_trade": 0.0, + "expectancy_pips": 0.0, + "total_return_pct": 0.0, + "cagr": 0.0, + "max_drawdown_pct": 0.0, + "sharpe": 0.0, + "sortino": 0.0, + "avg_trade_duration_bars": 0.0, + "longest_winning_streak": 0, + "longest_losing_streak": 0, + "final_balance": initial_balance, + } + + pnl_currency = np.array([t.pnl_currency for t in trades]) + pnl_pips = np.array([t.pnl_pips for t in trades]) + + wins = pnl_currency > 0 + losses = pnl_currency < 0 + + total_wins = pnl_currency[wins].sum() if wins.any() else 0.0 + total_losses = -pnl_currency[losses].sum() if losses.any() else 0.0 + + win_rate = wins.sum() / len(trades) * 100 + profit_factor = (total_wins / total_losses) if total_losses > 0 else float("inf") + expectancy_per_trade = pnl_currency.mean() + expectancy_pips = pnl_pips.mean() + + final_balance = float(equity_curve.iloc[-1]) if len(equity_curve) else initial_balance + total_return_pct = (final_balance / initial_balance - 1) * 100 + + # CAGR + if elapsed_days and elapsed_days > 0 and final_balance > 0: + years = elapsed_days / 365.25 + cagr = ((final_balance / initial_balance) ** (1 / years) - 1) * 100 if years > 0 else 0.0 + else: + cagr = 0.0 + + # Max drawdown + if len(equity_curve): + running_max = equity_curve.cummax() + drawdown = (equity_curve - running_max) / running_max + max_dd = float(drawdown.min()) * 100 + else: + max_dd = 0.0 + + # Sharpe / Sortino — daily returns from equity curve + sharpe, sortino = 0.0, 0.0 + if len(equity_curve) > 2: + daily = equity_curve.resample("1D").last().ffill().pct_change().dropna() + if len(daily) > 1 and daily.std() > 0: + sharpe = float(daily.mean() / daily.std() * np.sqrt(252)) + downside = daily[daily < 0] + if len(downside) > 1 and downside.std() > 0: + sortino = float(daily.mean() / downside.std() * np.sqrt(252)) + + # Streaks + win_loss_seq = [1 if t.pnl_currency > 0 else (-1 if t.pnl_currency < 0 else 0) for t in trades] + longest_win = _longest_streak(win_loss_seq, 1) + longest_loss = _longest_streak(win_loss_seq, -1) + + avg_duration = float(np.mean([t.duration_bars for t in trades])) + + return { + "total_trades": len(trades), + "wins": int(wins.sum()), + "losses": int(losses.sum()), + "win_rate": float(win_rate), + "profit_factor": float(profit_factor), + "expectancy_per_trade": float(expectancy_per_trade), + "expectancy_pips": float(expectancy_pips), + "total_return_pct": float(total_return_pct), + "cagr": float(cagr), + "max_drawdown_pct": float(max_dd), + "sharpe": float(sharpe), + "sortino": float(sortino), + "avg_trade_duration_bars": avg_duration, + "longest_winning_streak": longest_win, + "longest_losing_streak": longest_loss, + "final_balance": final_balance, + } + + +def _longest_streak(seq: List[int], target: int) -> int: + best = current = 0 + for x in seq: + if x == target: + current += 1 + best = max(best, current) + else: + current = 0 + return best + + +def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame: + """Convert trade journal to a DataFrame for CSV/JSON export.""" + if not trades: + return pd.DataFrame() + return pd.DataFrame([asdict(t) for t in trades]) diff --git a/news_engine.py b/news_engine.py index 7424c62..49ab66f 100644 --- a/news_engine.py +++ b/news_engine.py @@ -413,15 +413,7 @@ class NewsEngine: @staticmethod def _get_affected_pairs(currency: str) -> List[str]: - """Get all major pairs affected by a currency event.""" - pairs = { - "USD": ["EURUSD=X", "GBPUSD=X", "USDJPY=X", "USDCHF=X", "AUDUSD=X", "USDCAD=X", "NZDUSD=X"], - "EUR": ["EURUSD=X", "EURGBP=X", "EURJPY=X", "EURCHF=X"], - "GBP": ["GBPUSD=X", "EURGBP=X", "GBPJPY=X"], - "JPY": ["USDJPY=X", "EURJPY=X", "GBPJPY=X", "AUDJPY=X", "CADJPY=X"], - "CHF": ["USDCHF=X", "EURCHF=X"], - "AUD": ["AUDUSD=X", "AUDJPY=X"], - "CAD": ["USDCAD=X", "CADJPY=X"], - "NZD": ["NZDUSD=X"], - } - return pairs.get(currency, []) + """Map a currency event to USD/JPY (the only watched pair).""" + if currency in ("USD", "JPY"): + return ["USDJPY=X"] + return [] diff --git a/requirements.txt b/requirements.txt index 2379543..45a3851 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,5 @@ numpy>=1.24 pandas>=2.0 requests>=2.28 +pyarrow>=14.0 +tqdm>=4.66 diff --git a/signal_aggregator.py b/signal_aggregator.py index 4010e29..f5179c7 100644 --- a/signal_aggregator.py +++ b/signal_aggregator.py @@ -1,16 +1,10 @@ """ -Forex Signal Intelligence Platform - Signal Aggregator -======================================================== -Combines technical analysis across multiple timeframes with -fundamental/news context to produce final actionable signals. - -This is the brain — it decides what's actually worth paying attention to. - -Key principles: -1. Multi-timeframe agreement = higher confidence -2. High-impact news overrides technicals -3. Divergences between timeframes = caution -4. No signal is better than a bad signal +ForexBot2 — Signal Aggregator +============================== +Combines multi-timeframe technical analysis with optional news context +into a graded actionable signal. Same code path is used for live mode +and walk-forward backtesting; the only difference is whether the input +OHLCV is fetched live or sliced from a pre-loaded historical buffer. """ import logging @@ -20,10 +14,7 @@ from typing import Dict, List, Optional import pandas as pd -from config import ( - PlatformConfig, SignalStrength, Timeframe, - SYMBOL_DISPLAY, -) +from config import PlatformConfig, SignalStrength, Timeframe from data_provider import ForexDataProvider from engine import TechnicalEngine, TechnicalAnalysis from news_engine import NewsEngine, FundamentalContext @@ -33,10 +24,9 @@ logger = logging.getLogger(__name__) @dataclass class TradeSetup: - """A complete trade setup recommendation.""" pair: str display_name: str - direction: str # "BUY" or "SELL" + direction: str # "BUY" / "SELL" entry_price: float stop_loss: float take_profit_1: float @@ -49,47 +39,38 @@ class TradeSetup: @dataclass class SignalReport: - """Complete signal report for a single pair.""" symbol: str display_name: str timestamp: str - # Composite signals primary_signal: SignalStrength = SignalStrength.NEUTRAL primary_score: float = 0.0 primary_confidence: float = 0.0 - # Multi-timeframe agreement mtf_agreement: bool = False mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict) mtf_scores: Dict[str, float] = field(default_factory=dict) - # Technical details primary_analysis: Optional[TechnicalAnalysis] = None confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list) - # Fundamental context event_risk: str = "clear" event_description: str = "" news_sentiment: float = 0.0 news_description: str = "" - # Final verdict final_signal: SignalStrength = SignalStrength.NEUTRAL final_score: float = 0.0 final_confidence: float = 0.0 action_recommendation: str = "" - signal_quality: str = "" # "A+", "A", "B", "C", "D", "F" + signal_quality: str = "" # A+, A, B, C, D, F - # Trade setup (only for actionable signals) trade_setup: Optional[TradeSetup] = None - # Key reasons bullish_reasons: List[str] = field(default_factory=list) bearish_reasons: List[str] = field(default_factory=list) warnings: List[str] = field(default_factory=list) - # Current price info current_price: float = 0.0 spread_estimate: float = 0.0 market_regime: str = "" @@ -98,53 +79,67 @@ class SignalReport: class SignalAggregator: """ - Master signal aggregator. - - Orchestrates data fetching, technical analysis across timeframes, - fundamental analysis, and produces final graded signals. + Master aggregator. Designed so the backtester can drive it bar-by-bar + via `analyze_from_dataframes` without re-fetching from the API. """ def __init__(self, config: PlatformConfig): self.config = config self.data_provider = ForexDataProvider(config) self.tech_engine = TechnicalEngine(config.technical) - self.news_engine = NewsEngine(config.news) + self._news_engine: Optional[NewsEngine] = None # lazy - def analyze_all_pairs(self) -> List[SignalReport]: - """ - Analyze all watched pairs and return sorted signal reports. - Returns reports sorted by signal quality (best first). - """ - reports = [] - context = self.news_engine.get_fundamental_context() + def _get_news_engine(self) -> NewsEngine: + if self._news_engine is None: + self._news_engine = NewsEngine(self.config.news) + return self._news_engine - # Batch-prefetch all pairs × all needed timeframes (3 API calls instead of 21) - all_timeframes = ( - [self.config.primary_timeframe] + self.config.confirmation_timeframes - ) - self.data_provider.prefetch_batch(self.config.watch_pairs, all_timeframes) - - for symbol in self.config.watch_pairs: - try: - report = self.analyze_pair(symbol, context) - reports.append(report) - except Exception as e: - logger.error(f"Failed to analyze {symbol}: {e}") - continue - - # Sort by absolute score (strongest signals first) - reports.sort(key=lambda r: abs(r.final_score), reverse=True) - return reports + # ── Public: live mode entry point ────────────────────────────── def analyze_pair( self, - symbol: str, + symbol: Optional[str] = None, context: Optional[FundamentalContext] = None, ) -> SignalReport: - """Complete analysis of a single pair across all layers.""" - if context is None: - context = self.news_engine.get_fundamental_context() + """Live-mode entry: fetch all needed timeframes then delegate.""" + if symbol is None: + symbol = self.config.symbol + timeframes = [self.config.primary_timeframe] + list( + self.config.confirmation_timeframes + ) + + dfs: Dict[Timeframe, pd.DataFrame] = {} + for tf in timeframes: + try: + dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf) + except Exception as e: + logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") + + return self.analyze_from_dataframes( + symbol=symbol, + dfs=dfs, + context=context, + use_news=True, + ) + + # ── Public: bar-replayable entry point ───────────────────────── + + def analyze_from_dataframes( + self, + symbol: str, + dfs: Dict[Timeframe, pd.DataFrame], + context: Optional[FundamentalContext] = None, + use_news: bool = True, + ) -> SignalReport: + """ + Run the full analysis pipeline on already-fetched dataframes. + + `dfs` must include the primary timeframe and ideally the + confirmation timeframes. Each df is treated as the visible history + at decision time — the backtester is responsible for slicing so + no future bars leak in. + """ pair_info = self.data_provider.get_pair_info(symbol) display = pair_info["display"] pip_value = pair_info["pip_value"] @@ -155,33 +150,33 @@ class SignalAggregator: timestamp=str(datetime.now(timezone.utc)), ) - # ── 1. Primary Timeframe Analysis ────────────────────────── - try: - primary_df = self.data_provider.fetch_ohlcv( - symbol, self.config.primary_timeframe - ) - primary_analysis = self.tech_engine.analyze( - primary_df, symbol, self.config.primary_timeframe.value - ) - report.primary_analysis = primary_analysis - report.primary_signal = primary_analysis.composite_signal - report.primary_score = primary_analysis.composite_score - report.primary_confidence = primary_analysis.confidence - report.market_regime = primary_analysis.market_regime - report.volatility_regime = primary_analysis.volatility_regime - report.current_price = float(primary_df["Close"].iloc[-1]) - report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal - report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score - except Exception as e: - logger.error(f"Primary analysis failed for {symbol}: {e}") - report.warnings.append(f"Primary timeframe data unavailable: {e}") + # ── 1. Primary timeframe ─────────────────────────────────── + primary_df = dfs.get(self.config.primary_timeframe) + if primary_df is None or primary_df.empty: + report.warnings.append("Primary timeframe data unavailable") report.signal_quality = "F" return report - # ── 2. Confirmation Timeframe Analyses ───────────────────── + primary_analysis = self.tech_engine.analyze( + primary_df, symbol, self.config.primary_timeframe.value + ) + report.primary_analysis = primary_analysis + report.primary_signal = primary_analysis.composite_signal + report.primary_score = primary_analysis.composite_score + report.primary_confidence = primary_analysis.confidence + report.market_regime = primary_analysis.market_regime + report.volatility_regime = primary_analysis.volatility_regime + report.current_price = float(primary_df["Close"].iloc[-1]) + report.timestamp = str(primary_df.index[-1]) + report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal + report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score + + # ── 2. Confirmation timeframes ───────────────────────────── for tf in self.config.confirmation_timeframes: + cf_df = dfs.get(tf) + if cf_df is None or cf_df.empty: + continue try: - cf_df = self.data_provider.fetch_ohlcv(symbol, tf) cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value) report.confirmation_analyses.append(cf_analysis) report.mtf_signals[tf.value] = cf_analysis.composite_signal @@ -189,35 +184,51 @@ class SignalAggregator: except Exception as e: logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}") - # ── 3. Multi-Timeframe Agreement Check ───────────────────── + # ── 3. MTF agreement ─────────────────────────────────────── report.mtf_agreement = self._check_mtf_agreement(report) - # ── 4. Fundamental Context ───────────────────────────────── - event_risk, event_desc = self.news_engine.get_event_risk_level(symbol, context) - report.event_risk = event_risk - report.event_description = event_desc + # ── 4. Fundamental context (optional) ────────────────────── + if use_news: + news = self._get_news_engine() + if context is None: + context = news.get_fundamental_context() + event_risk, event_desc = news.get_event_risk_level(symbol, context) + report.event_risk = event_risk + report.event_description = event_desc + sentiment, sent_desc = news.get_pair_sentiment(symbol, context) + report.news_sentiment = sentiment + report.news_description = sent_desc + else: + report.event_risk = "clear" + report.event_description = "" + report.news_sentiment = 0.0 + report.news_description = "" - sentiment, sent_desc = self.news_engine.get_pair_sentiment(symbol, context) - report.news_sentiment = sentiment - report.news_description = sent_desc + # ── 5. Spread estimate (only meaningful in live mode) ────── + if use_news: + try: + report.spread_estimate = self.data_provider.get_spread_estimate(symbol) + except Exception: + report.spread_estimate = 0.0 + else: + # Backtest sets spread externally via BacktestConfig + report.spread_estimate = 0.0 - # ── 5. Spread Estimate ───────────────────────────────────── - report.spread_estimate = self.data_provider.get_spread_estimate(symbol) - - # ── 6. Compute Final Signal ──────────────────────────────── + # ── 6. Final signal ──────────────────────────────────────── self._compute_final_signal(report) - # ── 7. Generate Trade Setup (if signal is actionable) ────── + # ── 7. Trade setup for actionable signals ────────────────── if report.signal_quality in ("A+", "A", "B"): report.trade_setup = self._generate_trade_setup(report, pip_value) - # ── 8. Collect Reasons ───────────────────────────────────── + # ── 8. Reasons ───────────────────────────────────────────── self._collect_reasons(report) return report + # ── Internals ────────────────────────────────────────────────── + def _check_mtf_agreement(self, report: SignalReport) -> bool: - """Check if multiple timeframes agree on direction.""" signals = list(report.mtf_signals.values()) if len(signals) < 2: return False @@ -229,28 +240,17 @@ class SignalAggregator: return bullish >= threshold or bearish >= threshold def _compute_final_signal(self, report: SignalReport): - """ - Compute the final signal by combining all layers. - - Priority: - 1. Event risk can downgrade or block signals - 2. MTF agreement boosts confidence - 3. News sentiment adds a small bias - 4. Quality grade combines everything - """ score = report.primary_score confidence = report.primary_confidence - # ── MTF adjustment ───────────────────────────────────────── + # MTF if report.mtf_agreement: - # All timeframes agree — boost confidence and blend scores confidence = min(confidence + 15, 95) all_scores = list(report.mtf_scores.values()) if all_scores: mtf_avg = sum(all_scores) / len(all_scores) score = score * 0.6 + mtf_avg * 0.4 else: - # Timeframes disagree — reduce confidence confidence *= 0.7 if len(report.mtf_scores) >= 2: scores = list(report.mtf_scores.values()) @@ -260,35 +260,33 @@ class SignalAggregator: report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes") confidence *= 0.6 - # ── News sentiment adjustment ────────────────────────────── + # News sentiment if abs(report.news_sentiment) > 0.2: - # Sentiment aligns with technical if (report.news_sentiment > 0 and score > 0) or \ (report.news_sentiment < 0 and score < 0): confidence = min(confidence + 5, 95) - # Sentiment conflicts with technical elif (report.news_sentiment > 0 and score < -10) or \ (report.news_sentiment < 0 and score > 10): report.warnings.append("News sentiment CONFLICTS with technical signal") confidence *= 0.85 - # ── Event risk adjustment ────────────────────────────────── + # Event risk if report.event_risk == "danger": report.warnings.append(f"EVENT RISK: {report.event_description}") - confidence *= 0.3 # Heavily penalize — event will disrupt price action + confidence *= 0.3 report.action_recommendation = f"WAIT — {report.event_description}" elif report.event_risk == "caution": report.warnings.append(f"Upcoming event: {report.event_description}") confidence *= 0.7 - # ── Volatility regime adjustment ─────────────────────────── + # Volatility regime if report.volatility_regime == "extreme": report.warnings.append("EXTREME volatility — wider stops needed, reduced position size") confidence *= 0.8 elif report.volatility_regime == "low": report.warnings.append("Low volatility — potential for breakout, tight ranges") - # ── Final signal determination ───────────────────────────── + # Final signal if score >= 40 and confidence >= 65: report.final_signal = SignalStrength.STRONG_BUY elif score >= 15 and confidence >= 50: @@ -303,7 +301,7 @@ class SignalAggregator: report.final_score = score report.final_confidence = confidence - # ── Quality grade ────────────────────────────────────────── + # Quality grade abs_score = abs(score) if abs_score >= 50 and confidence >= 75 and report.mtf_agreement: report.signal_quality = "A+" @@ -318,8 +316,7 @@ class SignalAggregator: else: report.signal_quality = "F" - # ── Action recommendation ────────────────────────────────── - if not report.action_recommendation: # not already set by event risk + if not report.action_recommendation: sig_name = report.final_signal.name.replace("_", " ") if report.signal_quality in ("A+", "A"): report.action_recommendation = f"{sig_name} — High-quality setup, consider entry" @@ -331,35 +328,29 @@ class SignalAggregator: report.action_recommendation = "NO TRADE — No clear signal" def _generate_trade_setup(self, report: SignalReport, pip_value: float) -> Optional[TradeSetup]: - """Generate entry, SL, TP levels for actionable signals.""" if not report.primary_analysis: return None price = report.current_price atr = report.primary_analysis.key_levels.get("atr", 0) - if atr == 0: return None is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) rr = self.config.signals.default_rr_ratio + sl_mult = self.config.signals.atr_sl_multiplier - # SL = 1.5x ATR from entry (adaptive to volatility) - sl_distance = atr * 1.5 + sl_distance = atr * sl_mult tp1_distance = sl_distance * rr tp2_distance = sl_distance * rr * 1.5 if is_buy: - entry = price - sl = price - sl_distance - tp1 = price + tp1_distance - tp2 = price + tp2_distance + entry, sl = price, price - sl_distance + tp1, tp2 = price + tp1_distance, price + tp2_distance direction = "BUY" else: - entry = price - sl = price + sl_distance - tp1 = price - tp1_distance - tp2 = price - tp2_distance + entry, sl = price, price + sl_distance + tp1, tp2 = price - tp1_distance, price - tp2_distance direction = "SELL" risk_pips = sl_distance / pip_value @@ -380,21 +371,12 @@ class SignalAggregator: ) def _collect_reasons(self, report: SignalReport): - """ - Collect bullish/bearish reasons from all indicators. - - Each indicator's description is captured once via the indicators - dict. Pattern signals (candlestick) are already included in the - 'candlestick' indicator description, so we do NOT re-process - pattern_signals to avoid duplicate entries. - """ if not report.primary_analysis: return for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue - if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY): prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else "" report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}") @@ -402,7 +384,6 @@ class SignalAggregator: prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else "" report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}") - # Add news context if report.news_sentiment > 0.1: report.bullish_reasons.append(f"News: {report.news_description}") elif report.news_sentiment < -0.1: