Backtest tuning, data freshness guard, Telegram improvements

- Drop USD/JPY from scan pairs (zero signals across all timeframes)
- Disable divergence indicator (zero impact in 14-config lab test)
- Parallelize evaluate.py with ProcessPoolExecutor (16 cores)
- Add --pair CLI filter to evaluate.py for single-pair backtests
- Add data freshness guard: stale cache auto-invalidated, primary TF
  stale data suppresses signals to prevent trading on old data
- Add /cache command: shows fetch age and last bar per pair/TF
- Add /clearcache command: wipes in-memory cache for fresh API fetch
- Remove BB from WebSocket breach alerts (keep structural levels only)
- Show live prices only for WS-monitored pairs (EUR/USD, XAU/USD)
- Replace arrow emojis with text tags in indicator lines

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
harwinder
2026-05-12 22:56:55 -04:00
parent 6fe1a3e0f0
commit 1e5047e20f
6 changed files with 1148 additions and 320 deletions
+613
View File
@@ -0,0 +1,613 @@
#!/usr/bin/env python3
"""
ForexBot2 — Historical Signal Evaluation
=========================================
Walks the engine over the last 7 days of market data for all scan pairs,
records every signal and trade setup, then checks outcomes on M5 data.
GPU-accelerated outcome evaluation via CuPy (RTX 4070 SUPER).
Falls back to NumPy if CuPy/CUDA is unavailable.
Usage:
python backtest/evaluate.py # full 7-day eval, M15+H1
python backtest/evaluate.py --tf H1 # single timeframe
python backtest/evaluate.py --days 3 # shorter window
python backtest/evaluate.py --no-cache # re-download all data
"""
import argparse
import csv
import os
import sys
import time
from concurrent.futures import ProcessPoolExecutor, as_completed
from dataclasses import dataclass, field
from datetime import datetime, timezone, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple
import numpy as np
import pandas as pd
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from config import (
PlatformConfig, Timeframe, SignalStrength,
EURUSD, USDJPY, GBPUSD, XAUUSD, PAIR_INFO,
)
from data_provider import ForexDataProvider
from signal_aggregator import SignalAggregator
# ── GPU setup ────────────────────────────────────────────────────────
try:
if "CUDA_PATH" not in os.environ:
_nvidia_dir = os.path.join(
sys.prefix, "lib", "python3.12", "site-packages", "nvidia"
)
if os.path.isdir(_nvidia_dir):
os.environ["CUDA_PATH"] = _nvidia_dir
import cupy as cp
_test = cp.array([1.0, 2.0])
_ = cp.where(_test > 0.5)
GPU = True
GPU_NAME = cp.cuda.runtime.getDeviceProperties(0)["name"].decode()
print(f"GPU: {GPU_NAME} (CuPy {cp.__version__})")
del _test
except Exception:
import numpy as cp # type: ignore[no-redef]
GPU = False
print("GPU: not available — falling back to NumPy")
# ── Constants ────────────────────────────────────────────────────────
BACKTEST_DIR = Path(__file__).resolve().parent
DATA_DIR = BACKTEST_DIR / "data"
RESULTS_DIR = BACKTEST_DIR / "results"
SCAN_PAIRS = [EURUSD, GBPUSD, XAUUSD]
EVAL_TIMEFRAMES = [Timeframe.M15, Timeframe.H1]
TF_CONFIRMATIONS = {
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
ALL_TFS_NEEDED = {
Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4, Timeframe.D1,
}
OUTCOME_TF = Timeframe.M5
MIN_LOOKBACK_BARS = 250
@dataclass
class Trade:
pair: str
display: str
timeframe: str
entry_time: pd.Timestamp
direction: str
entry_price: float
stop_loss: float
take_profit_1: float
take_profit_2: Optional[float]
risk_reward: float
risk_pips: float
reward_pips: float
signal_quality: str
confidence: float
score: float
regime: str
outcome: str = "OPEN"
exit_price: float = 0.0
exit_time: Optional[pd.Timestamp] = None
pnl_pips: float = 0.0
bars_held: int = 0
# ══════════════════════════════════════════════════════════════════════
# DATA FETCHING & CACHING
# ══════════════════════════════════════════════════════════════════════
def _cache_path(symbol: str, tf: Timeframe) -> Path:
safe = symbol.replace("/", "_").replace("=", "")
return DATA_DIR / f"{safe}_{tf.value}.parquet"
def fetch_all_data(
pairs: List[str],
timeframes,
lookback_days: int = 67,
use_cache: bool = True,
) -> Dict[str, Dict[Timeframe, pd.DataFrame]]:
"""Fetch OHLCV for all pair×TF combos, cache to parquet."""
config = PlatformConfig()
config.data_lookback_days = lookback_days
dp = ForexDataProvider(config)
data: Dict[str, Dict[Timeframe, pd.DataFrame]] = {}
total = len(pairs) * len(timeframes)
done = 0
for symbol in pairs:
data[symbol] = {}
display = PAIR_INFO.get(symbol, {}).get("display", symbol)
for tf in timeframes:
done += 1
cache = _cache_path(symbol, tf)
if use_cache and cache.exists():
mtime = datetime.fromtimestamp(cache.stat().st_mtime, tz=timezone.utc)
age_hours = (datetime.now(timezone.utc) - mtime).total_seconds() / 3600
if age_hours < 4:
df = pd.read_parquet(cache)
data[symbol][tf] = df
print(f" [{done}/{total}] {display} {tf.value} — cached ({len(df)} bars)")
continue
print(f" [{done}/{total}] {display} {tf.value} — fetching...", end="", flush=True)
try:
df = dp.fetch_ohlcv(symbol, tf, lookback_days=lookback_days)
data[symbol][tf] = df
try:
df.to_parquet(cache)
except Exception:
pass
print(f" {len(df)} bars")
except Exception as e:
print(f" FAILED: {e}")
return data
# ══════════════════════════════════════════════════════════════════════
# WALK-FORWARD EVALUATION
# ══════════════════════════════════════════════════════════════════════
def walk_forward(
pair: str,
eval_tf: Timeframe,
all_data: Dict[Timeframe, pd.DataFrame],
eval_days: int,
) -> List[Trade]:
"""Walk through eval_days of candles, run the engine at each bar."""
primary_df = all_data.get(eval_tf)
if primary_df is None or primary_df.empty:
return []
conf_tfs = TF_CONFIRMATIONS.get(eval_tf, [Timeframe.M15, Timeframe.H4])
display = PAIR_INFO.get(pair, {}).get("display", pair)
cutoff = primary_df.index[-1] - pd.Timedelta(days=eval_days)
eval_indices = primary_df.index[primary_df.index >= cutoff]
if len(eval_indices) == 0:
return []
first_eval_pos = primary_df.index.get_loc(eval_indices[0])
if first_eval_pos < MIN_LOOKBACK_BARS:
first_eval_pos = MIN_LOOKBACK_BARS
eval_indices = primary_df.index[first_eval_pos:]
config = PlatformConfig()
config.primary_timeframe = eval_tf
config.confirmation_timeframes = conf_tfs
aggregator = SignalAggregator(config)
cooldown_bars = {
Timeframe.M5: 24, Timeframe.M15: 10,
Timeframe.H1: 5, Timeframe.H4: 3,
}.get(eval_tf, 6)
trades: List[Trade] = []
total = len(eval_indices)
bars_until_next = 0
for i, eval_time in enumerate(eval_indices):
if (i + 1) % max(1, total // 10) == 0:
print(f" {display} {eval_tf.value}: {i+1}/{total} candles...", flush=True)
if bars_until_next > 0:
bars_until_next -= 1
continue
dfs: Dict[Timeframe, pd.DataFrame] = {}
dfs[eval_tf] = primary_df.loc[:eval_time]
for ctf in conf_tfs:
ctf_df = all_data.get(ctf)
if ctf_df is not None and not ctf_df.empty:
dfs[ctf] = ctf_df.loc[:eval_time]
try:
report = aggregator.analyze_from_dataframes(
symbol=pair, dfs=dfs, use_news=False,
)
except Exception:
continue
if report.trade_setup and report.signal_quality in ("A+", "A", "B", "C"):
ts = report.trade_setup
trades.append(Trade(
pair=pair,
display=display,
timeframe=eval_tf.value,
entry_time=eval_time,
direction=ts.direction,
entry_price=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=ts.risk_pips,
reward_pips=ts.reward_pips,
signal_quality=report.signal_quality,
confidence=report.final_confidence,
score=report.final_score,
regime=report.market_regime,
))
bars_until_next = cooldown_bars
return trades
def _walk_forward_worker(args):
"""Top-level worker for multiprocessing (must be picklable)."""
pair, tf, pair_data, eval_days = args
display = PAIR_INFO.get(pair, {}).get("display", pair)
trades = walk_forward(pair, tf, pair_data, eval_days)
return pair, tf, display, trades
# ══════════════════════════════════════════════════════════════════════
# GPU-ACCELERATED OUTCOME EVALUATION
# ══════════════════════════════════════════════════════════════════════
def evaluate_outcomes(
trades: List[Trade],
m5_data: Dict[str, pd.DataFrame],
max_bars: int = 500,
) -> List[Trade]:
"""
For each trade, scan forward in M5 data to check if TP1 or SL hit first.
Uses GPU (CuPy) for vectorized parallel evaluation across all trades per pair.
"""
if not trades:
return trades
pairs_in_trades = set(t.pair for t in trades)
label = "GPU" if GPU else "CPU"
print(f"\nEvaluating {len(trades)} trade outcomes on {label}...")
for pair in pairs_in_trades:
df = m5_data.get(pair)
if df is None or df.empty:
continue
pair_trades = [t for t in trades if t.pair == pair]
if not pair_trades:
continue
highs_np = df["High"].values.astype(np.float64)
lows_np = df["Low"].values.astype(np.float64)
closes_np = df["Close"].values.astype(np.float64)
timestamps = df.index
highs_gpu = cp.asarray(highs_np)
lows_gpu = cp.asarray(lows_np)
for trade in pair_trades:
start_mask = timestamps > trade.entry_time
start_indices = np.where(start_mask)[0]
if len(start_indices) == 0:
continue
start = int(start_indices[0])
end = min(start + max_bars, len(highs_gpu))
if start >= end:
continue
h_slice = highs_gpu[start:end]
l_slice = lows_gpu[start:end]
if trade.direction == "BUY":
tp_mask = h_slice >= trade.take_profit_1
sl_mask = l_slice <= trade.stop_loss
else:
tp_mask = l_slice <= trade.take_profit_1
sl_mask = h_slice >= trade.stop_loss
tp_hits = cp.where(tp_mask)[0]
sl_hits = cp.where(sl_mask)[0]
tp_idx = int(tp_hits[0]) if len(tp_hits) > 0 else max_bars + 1
sl_idx = int(sl_hits[0]) if len(sl_hits) > 0 else max_bars + 1
pip_val = PAIR_INFO.get(pair, {}).get("pip_value", 0.0001)
if tp_idx <= sl_idx and tp_idx < max_bars + 1:
trade.outcome = "WIN"
trade.bars_held = tp_idx + 1
real_idx = start + tp_idx
trade.exit_price = trade.take_profit_1
trade.exit_time = timestamps[real_idx]
trade.pnl_pips = trade.reward_pips
elif sl_idx < tp_idx and sl_idx < max_bars + 1:
trade.outcome = "LOSS"
trade.bars_held = sl_idx + 1
real_idx = start + sl_idx
trade.exit_price = trade.stop_loss
trade.exit_time = timestamps[real_idx]
trade.pnl_pips = -trade.risk_pips
else:
trade.outcome = "OPEN"
last_idx = end - 1
trade.exit_time = timestamps[last_idx]
exit_p = float(closes_np[last_idx])
trade.exit_price = exit_p
if trade.direction == "BUY":
trade.pnl_pips = (exit_p - trade.entry_price) / pip_val
else:
trade.pnl_pips = (trade.entry_price - exit_p) / pip_val
trade.bars_held = end - start
return trades
# ══════════════════════════════════════════════════════════════════════
# REPORTING
# ══════════════════════════════════════════════════════════════════════
def print_report(trades: List[Trade], eval_days: int):
print("\n")
print("=" * 90)
print(f" FOREXBOT2 — ENGINE EVALUATION REPORT ({eval_days}-DAY)")
print(f" {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
print("=" * 90)
if not trades:
print("\n No trades generated.\n")
return
closed = [t for t in trades if t.outcome in ("WIN", "LOSS")]
wins = [t for t in closed if t.outcome == "WIN"]
losses = [t for t in closed if t.outcome == "LOSS"]
still_open = [t for t in trades if t.outcome == "OPEN"]
# ── Overall ──────────────────────────────────────────────────
print(f"\n OVERALL")
print(f" {'─' * 50}")
print(f" Total signals with trade setup: {len(trades)}")
print(f" Closed: {len(closed)} | Still open: {len(still_open)}")
if closed:
win_rate = len(wins) / len(closed) * 100
total_pnl = sum(t.pnl_pips for t in closed)
avg_win = np.mean([t.pnl_pips for t in wins]) if wins else 0
avg_loss = np.mean([abs(t.pnl_pips) for t in losses]) if losses else 0
gross_profit = sum(t.pnl_pips for t in wins) if wins else 0
gross_loss = sum(abs(t.pnl_pips) for t in losses) if losses else 0
profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf")
avg_rr = np.mean([t.risk_reward for t in closed])
avg_bars = np.mean([t.bars_held for t in closed])
print(f" Win rate: {win_rate:.1f}% ({len(wins)}W / {len(losses)}L)")
print(f" Profit factor: {profit_factor:.2f}")
print(f" Total P&L: {total_pnl:+.1f} pips")
print(f" Avg win: +{avg_win:.1f} pips | Avg loss: -{avg_loss:.1f} pips")
print(f" Avg R:R setup: 1:{avg_rr:.1f}")
print(f" Avg hold: {avg_bars:.0f} M5 bars ({avg_bars * 5:.0f} min)")
# Consecutive streaks
max_wins = max_losses = cur_wins = cur_losses = 0
for t in closed:
if t.outcome == "WIN":
cur_wins += 1
cur_losses = 0
max_wins = max(max_wins, cur_wins)
else:
cur_losses += 1
cur_wins = 0
max_losses = max(max_losses, cur_losses)
print(f" Max streak: {max_wins}W / {max_losses}L")
# ── By pair ──────────────────────────────────────────────────
print(f"\n BY PAIR")
print(f" {'─' * 50}")
print(f" {'Pair':<12} {'Trades':>7} {'Closed':>7} {'Win%':>6} {'P&L':>10} {'PF':>7}")
for pair in SCAN_PAIRS:
display = PAIR_INFO.get(pair, {}).get("display", pair)
pt = [t for t in trades if t.pair == pair]
pc = [t for t in pt if t.outcome in ("WIN", "LOSS")]
pw = [t for t in pc if t.outcome == "WIN"]
if not pt:
continue
wr = (len(pw) / len(pc) * 100) if pc else 0
pnl = sum(t.pnl_pips for t in pc)
gp = sum(t.pnl_pips for t in pw) if pw else 0
gl = sum(abs(t.pnl_pips) for t in pc if t.outcome == "LOSS") or 1
pf = gp / gl if gl > 0 else 0
print(f" {display:<12} {len(pt):>7} {len(pc):>7} {wr:>5.1f}% {pnl:>+9.1f}p {pf:>6.2f}")
# ── By timeframe ─────────────────────────────────────────────
print(f"\n BY TIMEFRAME")
print(f" {'─' * 50}")
print(f" {'TF':<8} {'Trades':>7} {'Closed':>7} {'Win%':>6} {'P&L':>10} {'PF':>7}")
tfs = sorted(set(t.timeframe for t in trades))
for tf in tfs:
tt = [t for t in trades if t.timeframe == tf]
tc = [t for t in tt if t.outcome in ("WIN", "LOSS")]
tw = [t for t in tc if t.outcome == "WIN"]
if not tt:
continue
wr = (len(tw) / len(tc) * 100) if tc else 0
pnl = sum(t.pnl_pips for t in tc)
gp = sum(t.pnl_pips for t in tw) if tw else 0
gl = sum(abs(t.pnl_pips) for t in tc if t.outcome == "LOSS") or 1
pf = gp / gl if gl > 0 else 0
print(f" {tf:<8} {len(tt):>7} {len(tc):>7} {wr:>5.1f}% {pnl:>+9.1f}p {pf:>6.2f}")
# ── By quality grade ─────────────────────────────────────────
print(f"\n BY QUALITY GRADE")
print(f" {'─' * 50}")
print(f" {'Grade':<8} {'Trades':>7} {'Closed':>7} {'Win%':>6} {'P&L':>10} {'Avg RR':>7}")
for grade in ("A+", "A", "B", "C"):
gt = [t for t in trades if t.signal_quality == grade]
gc = [t for t in gt if t.outcome in ("WIN", "LOSS")]
gw = [t for t in gc if t.outcome == "WIN"]
if not gt:
continue
wr = (len(gw) / len(gc) * 100) if gc else 0
pnl = sum(t.pnl_pips for t in gc)
arr = np.mean([t.risk_reward for t in gt])
print(f" {grade:<8} {len(gt):>7} {len(gc):>7} {wr:>5.1f}% {pnl:>+9.1f}p {arr:>6.1f}")
# ── By regime ────────────────────────────────────────────────
print(f"\n BY MARKET REGIME")
print(f" {'─' * 50}")
print(f" {'Regime':<16} {'Trades':>7} {'Win%':>6} {'P&L':>10}")
for regime in sorted(set(t.regime for t in trades)):
rt = [t for t in trades if t.regime == regime]
rc = [t for t in rt if t.outcome in ("WIN", "LOSS")]
rw = [t for t in rc if t.outcome == "WIN"]
wr = (len(rw) / len(rc) * 100) if rc else 0
pnl = sum(t.pnl_pips for t in rc)
label = regime.replace("_", " ").title()
print(f" {label:<16} {len(rt):>7} {wr:>5.1f}% {pnl:>+9.1f}p")
# ── Direction ────────────────────────────────────────────────
print(f"\n BY DIRECTION")
print(f" {'─' * 50}")
for d in ("BUY", "SELL"):
dt = [t for t in trades if t.direction == d]
dc = [t for t in dt if t.outcome in ("WIN", "LOSS")]
dw = [t for t in dc if t.outcome == "WIN"]
wr = (len(dw) / len(dc) * 100) if dc else 0
pnl = sum(t.pnl_pips for t in dc)
print(f" {d:<8} {len(dt):>7} trades {wr:>5.1f}% win {pnl:>+9.1f}p")
print(f"\n{'=' * 90}\n")
def save_trades_csv(trades: List[Trade], filename: str):
path = RESULTS_DIR / filename
with open(path, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow([
"pair", "display", "timeframe", "entry_time", "direction",
"entry_price", "stop_loss", "take_profit_1", "risk_reward",
"risk_pips", "reward_pips", "signal_quality", "confidence",
"score", "regime", "outcome", "exit_price", "exit_time",
"pnl_pips", "bars_held",
])
for t in trades:
writer.writerow([
t.pair, t.display, t.timeframe, t.entry_time, t.direction,
f"{t.entry_price:.5f}", f"{t.stop_loss:.5f}", f"{t.take_profit_1:.5f}",
f"{t.risk_reward:.2f}", f"{t.risk_pips:.1f}", f"{t.reward_pips:.1f}",
t.signal_quality, f"{t.confidence:.1f}", f"{t.score:.1f}",
t.regime, t.outcome, f"{t.exit_price:.5f}",
t.exit_time or "", f"{t.pnl_pips:.1f}", t.bars_held,
])
print(f"Trades saved to {path}")
# ══════════════════════════════════════════════════════════════════════
# MAIN
# ══════════════════════════════════════════════════════════════════════
def main():
parser = argparse.ArgumentParser(description="ForexBot2 signal engine evaluator")
parser.add_argument("--days", type=int, default=7, help="Evaluation window in days (default: 7)")
parser.add_argument("--tf", default=None, help="Single timeframe to evaluate (M15, H1)")
parser.add_argument("--no-cache", action="store_true", help="Re-download all data")
parser.add_argument("--max-outcome-bars", type=int, default=1000,
help="Max M5 bars to scan for SL/TP hit (default: 1000 = ~83h)")
parser.add_argument("--pair", default=None,
help="Single pair to evaluate (EURUSD, USDJPY, GBPUSD, XAUUSD)")
args = parser.parse_args()
eval_tfs = EVAL_TIMEFRAMES
if args.tf:
tf_map = {"M5": Timeframe.M5, "M15": Timeframe.M15, "H1": Timeframe.H1, "H4": Timeframe.H4}
if args.tf.upper() not in tf_map:
sys.exit(f"Invalid timeframe: {args.tf}. Choose from: {list(tf_map.keys())}")
eval_tfs = [tf_map[args.tf.upper()]]
pair_map = {"EURUSD": EURUSD, "USDJPY": USDJPY, "GBPUSD": GBPUSD, "XAUUSD": XAUUSD}
scan_pairs = SCAN_PAIRS
if args.pair:
key = args.pair.upper().replace("/", "")
if key not in pair_map:
sys.exit(f"Invalid pair: {args.pair}. Choose from: {list(pair_map.keys())}")
scan_pairs = [pair_map[key]]
DATA_DIR.mkdir(parents=True, exist_ok=True)
RESULTS_DIR.mkdir(parents=True, exist_ok=True)
# ── 1. Fetch data ────────────────────────────────────────────
lookback = args.days + 60
print(f"\n{'=' * 60}")
print(f" Fetching {lookback} days of data for {len(scan_pairs)} pair(s)...")
print(f"{'=' * 60}")
t0 = time.time()
all_data = fetch_all_data(
scan_pairs, ALL_TFS_NEEDED,
lookback_days=lookback,
use_cache=not args.no_cache,
)
print(f" Data fetched in {time.time() - t0:.1f}s")
# ── 2. Walk-forward evaluation (multi-core) ──────────────────
print(f"\n{'=' * 60}")
n_workers = min(os.cpu_count() or 4, 16)
print(f" Walk-forward evaluation: {args.days} days, TFs: {[tf.value for tf in eval_tfs]}")
print(f" Using {n_workers} CPU cores")
print(f"{'=' * 60}")
t1 = time.time()
jobs = []
for pair in scan_pairs:
pair_data = all_data.get(pair, {})
if not pair_data:
display = PAIR_INFO.get(pair, {}).get("display", pair)
print(f" {display}: no data, skipping")
continue
for tf in eval_tfs:
jobs.append((pair, tf, pair_data, args.days))
all_trades: List[Trade] = []
with ProcessPoolExecutor(max_workers=n_workers) as pool:
futures = {pool.submit(_walk_forward_worker, job): job for job in jobs}
for future in as_completed(futures):
pair, tf, display, trades = future.result()
print(f" {display} {tf.value}: {len(trades)} trade setups")
all_trades.extend(trades)
print(f"\n Walk-forward complete in {time.time() - t1:.1f}s")
print(f" Total trade setups: {len(all_trades)}")
# ── 3. Outcome evaluation (GPU) ──────────────────────────────
m5_data = {pair: all_data.get(pair, {}).get(Timeframe.M5) for pair in scan_pairs}
m5_data = {k: v for k, v in m5_data.items() if v is not None}
t2 = time.time()
all_trades = evaluate_outcomes(all_trades, m5_data, max_bars=args.max_outcome_bars)
print(f" Outcomes evaluated in {time.time() - t2:.3f}s")
# ── 4. Report ────────────────────────────────────────────────
print_report(all_trades, args.days)
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
save_trades_csv(all_trades, f"eval_{args.days}d_{ts}.csv")
if __name__ == "__main__":
main()
+15 -16
View File
@@ -76,25 +76,20 @@ class TechnicalConfig:
atr_period: int = 14 atr_period: int = 14
stoch_k: int = 14
stoch_d: int = 3
stoch_overbought: float = 80.0
stoch_oversold: float = 20.0
adx_period: int = 14 adx_period: int = 14
adx_strong_trend: float = 25.0 adx_strong_trend: float = 25.0
adx_very_strong: float = 50.0 adx_very_strong: float = 50.0
ichimoku_tenkan: int = 9
ichimoku_kijun: int = 26
ichimoku_senkou_b: int = 52
volume_ma_period: int = 20
pivot_lookback: int = 1
divergence_lookback: int = 14 divergence_lookback: int = 14
swing_sr_lookback: int = 50 swing_sr_lookback: int = 50
disabled_indicators: List[str] = field(default_factory=list) disabled_indicators: List[str] = field(default_factory=lambda: ["divergence"])
ema_mode: str = "4ema" # "4ema" (9/21/50/200) or "2ema" (50/200)
rsi_mode: str = "classic" # "classic" or "regime_aware"
macd_as_pillar: bool = False # promote MACD histogram to scored signal
regime_weights_override: Dict[str, Dict[str, float]] = field(default_factory=dict)
@dataclass @dataclass
@@ -106,8 +101,8 @@ class SignalConfig:
"ema_trend": 1.0, "ema_trend": 1.0,
"rsi": 1.5, "rsi": 1.5,
"bollinger": 1.5, "bollinger": 1.5,
"candlestick": 1.5, "candlestick": 1.0,
"divergence": 1.0, "divergence": 1.5,
}) })
mtf_agreement_threshold: int = 2 # of 3 timeframes mtf_agreement_threshold: int = 2 # of 3 timeframes
@@ -140,9 +135,13 @@ class FilterConfig:
regime_filter_timeframe: Timeframe = Timeframe.H4 regime_filter_timeframe: Timeframe = Timeframe.H4
regime_filter_min_adx: float = 25.0 regime_filter_min_adx: float = 25.0
session_filter: bool = False session_filter: bool = True
session_start_hour: int = 7 session_start_hour: int = 7
session_end_hour: int = 16 session_end_hour: int = 16
blocked_hours_utc: List[int] = field(default_factory=lambda: [0, 1, 2, 5, 6, 8, 12, 13, 14, 15, 16, 17, 21, 22, 23])
sell_filter: bool = True
sell_allowed_pairs: List[str] = field(default_factory=lambda: ["XAU/USD"])
@dataclass @dataclass
@@ -169,7 +168,7 @@ class PlatformConfig:
refresh_interval: int = 300 refresh_interval: int = 300
scan_pairs: List[str] = field( scan_pairs: List[str] = field(
default_factory=lambda: [EURUSD, USDJPY, GBPUSD, XAUUSD] default_factory=lambda: [EURUSD, GBPUSD, XAUUSD]
) )
scan_timeframes: List[Timeframe] = field( scan_timeframes: List[Timeframe] = field(
default_factory=lambda: [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4] default_factory=lambda: [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4]
+43 -2
View File
@@ -95,7 +95,11 @@ class ForexDataProvider:
): ):
cached = self._cache.get(cache_key) cached = self._cache.get(cache_key)
if cached is not None and not cached.empty: if cached is not None and not cached.empty:
return cached if self._is_data_fresh(cached, symbol, timeframe):
return cached
logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
del self._cache[cache_key]
del self._cache_timestamps[cache_key]
outputsize = max( outputsize = max(
300, 300,
@@ -104,10 +108,48 @@ class ForexDataProvider:
df = self._fetch_twelve_data(symbol, timeframe, outputsize) df = self._fetch_twelve_data(symbol, timeframe, outputsize)
df = self._validate_and_clean(df, symbol, timeframe) df = self._validate_and_clean(df, symbol, timeframe)
if not self._is_data_fresh(df, symbol, timeframe):
logger.warning(
f"STALE DATA from API: {symbol} {timeframe.value} — "
f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
)
self._cache[cache_key] = df self._cache[cache_key] = df
self._cache_timestamps[cache_key] = now self._cache_timestamps[cache_key] = now
return df return df
def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
"""Check if the last candle is within acceptable age for this timeframe."""
if df is None or df.empty:
return False
now = pd.Timestamp.now(tz="UTC")
if now.weekday() >= 5:
return True
last_bar = df.index[-1]
candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
max_age_seconds = candle_seconds * 3
age_seconds = (now - last_bar).total_seconds()
if age_seconds > max_age_seconds:
logger.warning(
f"{symbol} {timeframe.value}: last bar {last_bar.strftime('%Y-%m-%d %H:%M')} "
f"is {age_seconds / 60:.0f}m old (limit: {max_age_seconds / 60:.0f}m)"
)
return False
return True
def clear_cache(self) -> int:
"""Wipe all cached data. Returns number of entries cleared."""
count = len(self._cache)
self._cache.clear()
self._cache_timestamps.clear()
logger.info(f"Cache cleared: {count} entries wiped")
return count
def get_latest_price(self, symbol: str = USDJPY) -> Optional[float]: def get_latest_price(self, symbol: str = USDJPY) -> Optional[float]:
try: try:
df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1) df = self.fetch_ohlcv(symbol, Timeframe.M5, lookback_days=1)
@@ -255,7 +297,6 @@ class ForexDataProvider:
min_required = max( min_required = max(
self.config.technical.ema_trend, self.config.technical.ema_trend,
self.config.technical.ichimoku_senkou_b,
200, 200,
) + 10 ) + 10
+261 -262
View File
@@ -91,7 +91,7 @@ class TechnicalEngine:
if "adx" not in disabled: self._compute_adx(df, result) if "adx" not in disabled: self._compute_adx(df, result)
if "atr" not in disabled: self._compute_atr(df, result) if "atr" not in disabled: self._compute_atr(df, result)
self._compute_macd_internal(df, result) self._compute_macd(df, result)
if "divergence" not in disabled: self._detect_divergences(df, result) if "divergence" not in disabled: self._detect_divergences(df, result)
if "structure" not in disabled: self._compute_structure_sr(df, result) if "structure" not in disabled: self._compute_structure_sr(df, result)
@@ -100,6 +100,10 @@ class TechnicalEngine:
# ── Classify market regime ───────────────────────────────── # ── Classify market regime ─────────────────────────────────
self._classify_regime(df, result) self._classify_regime(df, result)
# ── Regime-aware RSI reinterpretation ──────────────────────
if self.config.rsi_mode == "regime_aware" and "rsi" not in disabled:
self._reinterpret_rsi_for_regime(result)
# ── Compute composite signal ─────────────────────────────── # ── Compute composite signal ───────────────────────────────
self._compute_composite(result) self._compute_composite(result)
@@ -110,7 +114,87 @@ class TechnicalEngine:
# ══════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════
def _compute_ema_signals(self, df: pd.DataFrame, result: TechnicalAnalysis): def _compute_ema_signals(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""EMA trend alignment analysis.""" """EMA trend alignment — supports 4-EMA and 2-EMA modes."""
if self.config.ema_mode == "2ema":
self._compute_ema_2(df, result)
else:
self._compute_ema_4(df, result)
def _compute_ema_2(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""EMA 50/200 trend alignment — cleaner, less noise on H1."""
c = self.config
close = df["Close"]
ema50 = close.ewm(span=c.ema_slow, adjust=False).mean()
ema200 = close.ewm(span=c.ema_trend, adjust=False).mean() if len(close) >= c.ema_trend else None
ema21 = close.ewm(span=c.ema_medium, adjust=False).mean()
latest = close.iloc[-1]
s = ema50.iloc[-1]
m = ema21.iloc[-1]
t = ema200.iloc[-1] if ema200 is not None and len(ema200) >= c.ema_trend else None
if t is None:
result.indicators["ema_trend"] = IndicatorSignal(
name="EMA Trend", value=0, signal=SignalStrength.NEUTRAL,
confidence=30, description="EMA200 data insufficient", weight=1.0,
)
result.key_levels["ema_medium"] = float(m)
result.key_levels["ema_slow"] = float(s)
return
# 3-level alignment: price vs EMA50, EMA50 vs EMA200
if latest > s > t:
signal = SignalStrength.STRONG_BUY
conf = 85
desc = "Full bullish: price > EMA50 > EMA200"
elif latest > s and s < t:
signal = SignalStrength.BUY
conf = 60
desc = "Emerging uptrend: price > EMA50, EMA50 < EMA200"
elif latest < s < t:
signal = SignalStrength.STRONG_SELL
conf = 85
desc = "Full bearish: price < EMA50 < EMA200"
elif latest < s and s > t:
signal = SignalStrength.SELL
conf = 60
desc = "Emerging downtrend: price < EMA50, EMA50 > EMA200"
else:
signal = SignalStrength.NEUTRAL
conf = 40
desc = "EMA alignment mixed"
# EMA50/200 crossover detection (recent 10 bars)
if len(ema50) >= 10 and ema200 is not None:
for lookback in range(2, min(11, len(ema50))):
prev_s = ema50.iloc[-lookback]
prev_t = ema200.iloc[-lookback]
if prev_s <= prev_t and s > t:
desc += " | Golden cross (50/200)"
conf = min(conf + 10, 95)
break
elif prev_s >= prev_t and s < t:
desc += " | Death cross (50/200)"
conf = min(conf + 10, 95)
break
normalized = 1.0 if signal == SignalStrength.STRONG_BUY else (
0.5 if signal == SignalStrength.BUY else (
-1.0 if signal == SignalStrength.STRONG_SELL else (
-0.5 if signal == SignalStrength.SELL else 0.0)))
result.indicators["ema_trend"] = IndicatorSignal(
name="EMA Trend", value=normalized,
signal=signal, confidence=conf,
description=desc, weight=1.0,
)
result.key_levels["ema_trend"] = float(t)
result.key_levels["ema_medium"] = float(m)
result.key_levels["ema_slow"] = float(s)
def _compute_ema_4(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Original 4-EMA (9/21/50/200) trend alignment analysis."""
c = self.config c = self.config
close = df["Close"] close = df["Close"]
@@ -123,8 +207,6 @@ class TechnicalEngine:
f, m, s = ema_fast.iloc[-1], ema_medium.iloc[-1], ema_slow.iloc[-1] f, m, s = ema_fast.iloc[-1], ema_medium.iloc[-1], ema_slow.iloc[-1]
t = ema_trend.iloc[-1] if ema_trend is not None and len(ema_trend) >= c.ema_trend else None t = ema_trend.iloc[-1] if ema_trend is not None and len(ema_trend) >= c.ema_trend else None
# Bullish: price > fast > medium > slow > trend
# Bearish: price < fast < medium < slow < trend
alignment_score = 0 alignment_score = 0
checks = [(latest, f), (f, m), (m, s)] checks = [(latest, f), (f, m), (m, s)]
if t is not None: if t is not None:
@@ -155,7 +237,6 @@ class TechnicalEngine:
signal = SignalStrength.NEUTRAL signal = SignalStrength.NEUTRAL
conf = 40 conf = 40
# Check for EMA crossover (recent)
crossover_desc = "" crossover_desc = ""
if len(ema_fast) >= 3: if len(ema_fast) >= 3:
prev_f, prev_m = ema_fast.iloc[-2], ema_medium.iloc[-2] prev_f, prev_m = ema_fast.iloc[-2], ema_medium.iloc[-2]
@@ -187,10 +268,10 @@ class TechnicalEngine:
gain = delta.where(delta > 0, 0.0) gain = delta.where(delta > 0, 0.0)
loss = (-delta).where(delta < 0, 0.0) loss = (-delta).where(delta < 0, 0.0)
avg_gain = gain.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period).mean() avg_gain = gain.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period, adjust=False).mean()
avg_loss = loss.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period).mean() avg_loss = loss.ewm(alpha=1/c.rsi_period, min_periods=c.rsi_period, adjust=False).mean()
rs = avg_gain / avg_loss.replace(0, np.nan) rs = avg_gain / avg_loss.replace(0, 1e-10)
rsi = 100 - (100 / (1 + rs)) rsi = 100 - (100 / (1 + rs))
rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0 rsi_val = rsi.iloc[-1] if not rsi.empty else 50.0
@@ -227,8 +308,8 @@ class TechnicalEngine:
# Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix) # Attach RSI series for divergence detection (dynamic attribute, filtered by _prefix)
result._rsi_series = rsi # type: ignore[attr-defined] result._rsi_series = rsi # type: ignore[attr-defined]
def _compute_macd_internal(self, df: pd.DataFrame, result: TechnicalAnalysis): def _compute_macd(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Compute MACD histogram internally for divergence detection (not a signal pillar).""" """MACD computation — always runs for divergence; optionally a signal pillar."""
c = self.config c = self.config
close = df["Close"] close = df["Close"]
@@ -240,13 +321,70 @@ class TechnicalEngine:
result._macd_histogram = histogram # type: ignore[attr-defined] result._macd_histogram = histogram # type: ignore[attr-defined]
if not c.macd_as_pillar:
return
if len(histogram) < 3:
return
hist_now = histogram.iloc[-1]
hist_prev = histogram.iloc[-2]
hist_prev2 = histogram.iloc[-3]
if pd.isna(hist_now) or pd.isna(hist_prev):
return
# Histogram zero-line crossover
crossed_up = hist_prev <= 0 < hist_now
crossed_down = hist_prev >= 0 > hist_now
# Momentum acceleration
accel_up = hist_now > 0 and hist_now > hist_prev > 0
accel_down = hist_now < 0 and hist_now < hist_prev < 0
# Momentum fading
fading_up = hist_now > 0 and hist_now < hist_prev
fading_down = hist_now < 0 and hist_now > hist_prev
# Normalize histogram relative to recent range for confidence
hist_abs = histogram.abs()
if len(hist_abs) >= 20:
hist_pct = (abs(hist_now) / hist_abs.iloc[-20:].mean()) if hist_abs.iloc[-20:].mean() > 0 else 1.0
else:
hist_pct = 1.0
if crossed_up or accel_up:
signal = SignalStrength.STRONG_BUY if accel_up else SignalStrength.BUY
conf = min(80, 55 + hist_pct * 10)
desc = "MACD histogram" + (" accelerating ↑" if accel_up else " crossed above zero")
elif crossed_down or accel_down:
signal = SignalStrength.STRONG_SELL if accel_down else SignalStrength.SELL
conf = min(80, 55 + hist_pct * 10)
desc = "MACD histogram" + (" accelerating ↓" if accel_down else " crossed below zero")
elif fading_up:
signal = SignalStrength.NEUTRAL
conf = 35
desc = f"MACD histogram positive but fading"
elif fading_down:
signal = SignalStrength.NEUTRAL
conf = 35
desc = f"MACD histogram negative but fading"
else:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"MACD histogram flat"
result.indicators["macd"] = IndicatorSignal(
name="MACD", value=float(hist_now),
signal=signal, confidence=conf,
description=desc, weight=1.5,
)
def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis): def _compute_bollinger(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Bollinger Bands position and squeeze detection.""" """Bollinger Bands position and squeeze detection."""
c = self.config c = self.config
close = df["Close"] close = df["Close"]
sma = close.rolling(c.bb_period).mean() sma = close.rolling(c.bb_period).mean()
std = close.rolling(c.bb_period).std() std = close.rolling(c.bb_period).std(ddof=0)
upper = sma + c.bb_std * std upper = sma + c.bb_std * std
lower = sma - c.bb_std * std lower = sma - c.bb_std * std
@@ -308,67 +446,6 @@ class TechnicalEngine:
result.key_levels["bb_lower"] = lower_val result.key_levels["bb_lower"] = lower_val
result.key_levels["bb_mid"] = sma_val result.key_levels["bb_mid"] = sma_val
def _compute_stochastic(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Stochastic oscillator with crossover signals."""
c = self.config
high = df["High"]
low = df["Low"]
close = df["Close"]
lowest_low = low.rolling(c.stoch_k).min()
highest_high = high.rolling(c.stoch_k).max()
denom = highest_high - lowest_low
denom = denom.replace(0, np.nan)
k_line = 100 * (close - lowest_low) / denom
d_line = k_line.rolling(c.stoch_d).mean()
k_val = k_line.iloc[-1] if not k_line.empty else 50.0
d_val = d_line.iloc[-1] if not d_line.empty else 50.0
if pd.isna(k_val):
k_val = 50.0
if pd.isna(d_val):
d_val = 50.0
# Crossover detection
crossover = ""
if len(k_line) >= 2 and len(d_line) >= 2:
prev_k = k_line.iloc[-2]
prev_d = d_line.iloc[-2]
if not pd.isna(prev_k) and not pd.isna(prev_d):
if prev_k <= prev_d and k_val > d_val:
crossover = " | Bullish %K/%D cross"
elif prev_k >= prev_d and k_val < d_val:
crossover = " | Bearish %K/%D cross"
if k_val >= c.stoch_overbought and d_val >= c.stoch_overbought:
signal = SignalStrength.SELL
conf = 60
desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Overbought{crossover}"
elif k_val <= c.stoch_oversold and d_val <= c.stoch_oversold:
signal = SignalStrength.BUY
conf = 60
desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Oversold{crossover}"
elif crossover and "Bullish" in crossover:
signal = SignalStrength.BUY
conf = 55
desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f}{crossover}"
elif crossover and "Bearish" in crossover:
signal = SignalStrength.SELL
conf = 55
desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f}{crossover}"
else:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"Stoch %K={k_val:.0f} %D={d_val:.0f} — Neutral"
result.indicators["stochastic"] = IndicatorSignal(
name="Stochastic", value=k_val, signal=signal,
confidence=conf, description=desc, weight=2.5,
)
def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis): def _compute_adx(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""ADX trend strength measurement.""" """ADX trend strength measurement."""
c = self.config c = self.config
@@ -389,12 +466,12 @@ class TechnicalEngine:
plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) 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) minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
atr = tr.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() atr = tr.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean()
plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() / atr) plus_di = 100 * (plus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr)
minus_di = 100 * (minus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() / atr) minus_di = 100 * (minus_dm.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean() / atr)
dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan) dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di).replace(0, np.nan)
adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period).mean() adx = dx.ewm(alpha=1/c.adx_period, min_periods=c.adx_period, adjust=False).mean()
adx_val = adx.iloc[-1] if not adx.empty and not pd.isna(adx.iloc[-1]) else 0.0 adx_val = adx.iloc[-1] if not adx.empty and not pd.isna(adx.iloc[-1]) else 0.0
plus_di_last = plus_di.iloc[-1] if not plus_di.empty else np.nan plus_di_last = plus_di.iloc[-1] if not plus_di.empty else np.nan
@@ -439,7 +516,7 @@ class TechnicalEngine:
tr3 = (low - close.shift()).abs() tr3 = (low - close.shift()).abs()
tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1) tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
atr = tr.ewm(alpha=1/c.atr_period, min_periods=c.atr_period).mean() atr = tr.ewm(alpha=1/c.atr_period, min_periods=c.atr_period, adjust=False).mean()
atr_val = atr.iloc[-1] if not atr.empty else 0.0 atr_val = atr.iloc[-1] if not atr.empty else 0.0
if pd.isna(atr_val): if pd.isna(atr_val):
atr_val = 0.0 atr_val = 0.0
@@ -473,91 +550,6 @@ class TechnicalEngine:
weight=0.0, # ATR is context, not a directional signal weight=0.0, # ATR is context, not a directional signal
) )
def _compute_ichimoku(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Ichimoku Cloud analysis."""
c = self.config
high, low, close = df["High"], df["Low"], df["Close"]
if len(df) < c.ichimoku_senkou_b + 26:
return
# Tenkan-sen (Conversion Line)
tenkan = (high.rolling(c.ichimoku_tenkan).max() +
low.rolling(c.ichimoku_tenkan).min()) / 2
# Kijun-sen (Base Line)
kijun = (high.rolling(c.ichimoku_kijun).max() +
low.rolling(c.ichimoku_kijun).min()) / 2
latest = close.iloc[-1]
tenkan_val = tenkan.iloc[-1]
kijun_val = kijun.iloc[-1]
# Current (unshifted) cloud
sa = (tenkan.iloc[-1] + kijun.iloc[-1]) / 2
sb_raw = (high.rolling(c.ichimoku_senkou_b).max() +
low.rolling(c.ichimoku_senkou_b).min()) / 2
sb = sb_raw.iloc[-1]
if pd.isna(sa) or pd.isna(sb) or pd.isna(tenkan_val) or pd.isna(kijun_val):
return
cloud_top = max(sa, sb)
cloud_bottom = min(sa, sb)
cloud_is_green = sa > sb
signals = []
score = 0
if latest > cloud_top:
signals.append("Price ABOVE cloud")
score += 2
elif latest < cloud_bottom:
signals.append("Price BELOW cloud")
score -= 2
else:
signals.append("Price IN cloud (indecision)")
if tenkan_val > kijun_val:
signals.append("TK cross bullish")
score += 1
elif tenkan_val < kijun_val:
signals.append("TK cross bearish")
score -= 1
if cloud_is_green:
signals.append("Green cloud")
score += 1
else:
signals.append("Red cloud")
score -= 1
if score >= 3:
signal = SignalStrength.STRONG_BUY
conf = 75
elif score >= 1:
signal = SignalStrength.BUY
conf = 55
elif score <= -3:
signal = SignalStrength.STRONG_SELL
conf = 75
elif score <= -1:
signal = SignalStrength.SELL
conf = 55
else:
signal = SignalStrength.NEUTRAL
conf = 35
result.indicators["ichimoku"] = IndicatorSignal(
name="Ichimoku", value=float(score), signal=signal,
confidence=conf,
description=f"Ichimoku [{score}/4]: {' | '.join(signals)}",
weight=0.3,
)
result.key_levels["ichimoku_cloud_top"] = cloud_top
result.key_levels["ichimoku_cloud_bottom"] = cloud_bottom
def _detect_divergences(self, df: pd.DataFrame, result: TechnicalAnalysis): def _detect_divergences(self, df: pd.DataFrame, result: TechnicalAnalysis):
""" """
Detect RSI and MACD divergences (powerful reversal signals). Detect RSI and MACD divergences (powerful reversal signals).
@@ -615,7 +607,7 @@ class TechnicalEngine:
name="Divergence", value=len(divergences), name="Divergence", value=len(divergences),
signal=signal, confidence=70, signal=signal, confidence=70,
description=" | ".join(divergences), description=" | ".join(divergences),
weight=1.0, weight=1.5,
) )
# Note: divergences are NOT added to pattern_signals to avoid # Note: divergences are NOT added to pattern_signals to avoid
# duplicate reason collection (the indicator entry covers them). # duplicate reason collection (the indicator entry covers them).
@@ -735,90 +727,96 @@ class TechnicalEngine:
name="Candlestick", value=float(bullish - bearish), name="Candlestick", value=float(bullish - bearish),
signal=signal, confidence=conf, signal=signal, confidence=conf,
description=" | ".join(patterns), description=" | ".join(patterns),
weight=1.5, weight=1.0,
) )
result.pattern_signals.extend(patterns) result.pattern_signals.extend(patterns)
def _compute_support_resistance(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Calculate pivot points and key S/R levels."""
if len(df) < 2:
return
# Use previous bar for pivot calculation
prev_h = df["High"].iloc[-2]
prev_l = df["Low"].iloc[-2]
prev_c = df["Close"].iloc[-2]
pivot = (prev_h + prev_l + prev_c) / 3
r1 = 2 * pivot - prev_l
s1 = 2 * pivot - prev_h
r2 = pivot + (prev_h - prev_l)
s2 = pivot - (prev_h - prev_l)
r3 = prev_h + 2 * (pivot - prev_l)
s3 = prev_l - 2 * (prev_h - pivot)
result.key_levels.update({
"pivot": pivot,
"r1": r1, "r2": r2, "r3": r3,
"s1": s1, "s2": s2, "s3": s3,
})
# Signal based on price position relative to pivot
latest = df["Close"].iloc[-1]
if latest > r1:
desc = f"Price above R1 ({r1:.5f}), next target R2 ({r2:.5f})"
elif latest > pivot:
desc = f"Price above Pivot ({pivot:.5f}), testing R1 ({r1:.5f})"
elif latest > s1:
desc = f"Price below Pivot ({pivot:.5f}), support at S1 ({s1:.5f})"
else:
desc = f"Price below S1 ({s1:.5f}), support at S2 ({s2:.5f})"
result.indicators["pivot"] = IndicatorSignal(
name="Pivot Points", value=pivot,
signal=SignalStrength.BUY if latest > pivot else SignalStrength.SELL,
confidence=45, description=desc, weight=0.8,
)
def _compute_volume_analysis(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Volume analysis (limited on Forex but still useful as a filter)."""
c = self.config
vol = df.get("Volume")
if vol is None or vol.sum() == 0:
return
vol_ma = vol.rolling(c.volume_ma_period).mean()
current_vol = vol.iloc[-1]
avg_vol = vol_ma.iloc[-1]
if pd.isna(avg_vol) or avg_vol == 0:
return
vol_ratio = current_vol / avg_vol
if vol_ratio > 2.0:
desc = f"Volume {vol_ratio:.1f}x average — HIGH ACTIVITY"
conf = 55
elif vol_ratio > 1.3:
desc = f"Volume {vol_ratio:.1f}x average — above normal"
conf = 45
elif vol_ratio < 0.5:
desc = f"Volume {vol_ratio:.1f}x average — LOW (signals less reliable)"
conf = 25
else:
desc = f"Volume {vol_ratio:.1f}x average — normal"
conf = 35
result.indicators["volume"] = IndicatorSignal(
name="Volume", value=vol_ratio,
signal=SignalStrength.NEUTRAL,
confidence=conf, description=desc, weight=0.0,
)
# ══════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════
# MARKET REGIME & COMPOSITE SCORING # MARKET REGIME & COMPOSITE SCORING
# ══════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════
def _reinterpret_rsi_for_regime(self, result: TechnicalAnalysis):
"""Reinterpret RSI signal based on market regime — called after regime classification."""
rsi_ind = result.indicators.get("rsi")
if not rsi_ind:
return
rsi_val = rsi_ind.value
regime = result.market_regime
if regime in ("trending_up", "trending"):
if 38 <= rsi_val <= 55:
signal = SignalStrength.STRONG_BUY
conf = 75
desc = f"RSI {rsi_val:.1f} — Pullback zone in uptrend (ideal entry)"
elif 30 <= rsi_val < 38:
signal = SignalStrength.BUY
conf = 60
desc = f"RSI {rsi_val:.1f} — Deep pullback in uptrend"
elif 55 < rsi_val <= 65:
signal = SignalStrength.BUY
conf = 50
desc = f"RSI {rsi_val:.1f} — Momentum ok in uptrend"
elif rsi_val > 65:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"RSI {rsi_val:.1f} — Extended in uptrend, wait for pullback"
elif rsi_val < 30:
signal = SignalStrength.NEUTRAL
conf = 25
desc = f"RSI {rsi_val:.1f} — Oversold in uptrend, possible reversal"
else:
return
elif regime == "trending_down":
if 45 <= rsi_val <= 62:
signal = SignalStrength.STRONG_SELL
conf = 75
desc = f"RSI {rsi_val:.1f} — Bounce zone in downtrend (ideal short)"
elif 62 < rsi_val <= 70:
signal = SignalStrength.SELL
conf = 60
desc = f"RSI {rsi_val:.1f} — Deep bounce in downtrend"
elif 35 <= rsi_val < 45:
signal = SignalStrength.SELL
conf = 50
desc = f"RSI {rsi_val:.1f} — Momentum ok in downtrend"
elif rsi_val < 35:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"RSI {rsi_val:.1f} — Extended in downtrend, wait for bounce"
elif rsi_val > 70:
signal = SignalStrength.NEUTRAL
conf = 25
desc = f"RSI {rsi_val:.1f} — Overbought in downtrend, possible reversal"
else:
return
else: # ranging
if rsi_val >= 70:
signal = SignalStrength.SELL
conf = 70
desc = f"RSI {rsi_val:.1f} — Overbought in range"
elif rsi_val <= 30:
signal = SignalStrength.BUY
conf = 70
desc = f"RSI {rsi_val:.1f} — Oversold in range"
elif rsi_val >= 60:
signal = SignalStrength.SELL
conf = 45
desc = f"RSI {rsi_val:.1f} — Upper range zone"
elif rsi_val <= 40:
signal = SignalStrength.BUY
conf = 45
desc = f"RSI {rsi_val:.1f} — Lower range zone"
else:
signal = SignalStrength.NEUTRAL
conf = 30
desc = f"RSI {rsi_val:.1f} — Mid-range, no edge"
result.indicators["rsi"] = IndicatorSignal(
name="RSI", value=rsi_val, signal=signal,
confidence=conf, description=desc, weight=rsi_ind.weight,
)
def _classify_regime(self, df: pd.DataFrame, result: TechnicalAnalysis): def _classify_regime(self, df: pd.DataFrame, result: TechnicalAnalysis):
"""Classify the market regime (trending vs ranging).""" """Classify the market regime (trending vs ranging)."""
adx_sig = result.indicators.get("adx") adx_sig = result.indicators.get("adx")
@@ -840,26 +838,29 @@ class TechnicalEngine:
Weighted average of all indicator signals, adjusted for: Weighted average of all indicator signals, adjusted for:
- Market regime (trend indicators weighted more in trends) - Market regime (trend indicators weighted more in trends)
- Divergence boost (divergences override in ranging markets)
- Confidence scaling - Confidence scaling
""" """
total_score = 0.0 total_score = 0.0
total_weight = 0.0 total_weight = 0.0
total_confidence = 0.0 total_confidence = 0.0
_trending = { # Use override weights if provided, else defaults
"ema_trend": 2.0, "rsi": 1.5, "bollinger": 0.8, if self.config.regime_weights_override:
"candlestick": 1.5, "divergence": 1.0, regime_weights = self.config.regime_weights_override
} else:
regime_weights = { _trending = {
"trending_up": _trending, "ema_trend": 2.0, "rsi": 1.5, "bollinger": 0.8,
"trending_down": _trending, "candlestick": 1.0, "divergence": 1.5,
"trending": _trending, }
"ranging": { regime_weights = {
"ema_trend": 0.5, "rsi": 2.0, "bollinger": 2.0, "trending_up": _trending,
"candlestick": 1.5, "divergence": 1.5, "trending_down": _trending,
}, "trending": _trending,
} "ranging": {
"ema_trend": 0.5, "rsi": 2.0, "bollinger": 2.0,
"candlestick": 1.0, "divergence": 2.0,
},
}
active_regime_weights = regime_weights.get(result.market_regime, {}) active_regime_weights = regime_weights.get(result.market_regime, {})
@@ -871,7 +872,7 @@ class TechnicalEngine:
regime_mult = active_regime_weights.get(name, 1.0) regime_mult = active_regime_weights.get(name, 1.0)
weight *= regime_mult weight *= regime_mult
signal_score = ind.signal.value # -2 to +2 signal_score = ind.signal.value
weighted_score = signal_score * weight * (ind.confidence / 100.0) weighted_score = signal_score * weight * (ind.confidence / 100.0)
total_score += weighted_score total_score += weighted_score
@@ -879,23 +880,21 @@ class TechnicalEngine:
total_confidence += ind.confidence * weight total_confidence += ind.confidence * weight
if total_weight > 0: if total_weight > 0:
normalized_score = (total_score / total_weight) * 50 # scale to -100..+100 normalized_score = (total_score / total_weight) * 50
avg_confidence = total_confidence / total_weight avg_confidence = total_confidence / total_weight
else: else:
normalized_score = 0 normalized_score = 0
avg_confidence = 0 avg_confidence = 0
# Clamp
normalized_score = max(-100, min(100, normalized_score)) normalized_score = max(-100, min(100, normalized_score))
# Determine composite signal if normalized_score >= 45:
if normalized_score >= 40:
result.composite_signal = SignalStrength.STRONG_BUY result.composite_signal = SignalStrength.STRONG_BUY
elif normalized_score >= 15: elif normalized_score >= 20:
result.composite_signal = SignalStrength.BUY result.composite_signal = SignalStrength.BUY
elif normalized_score <= -40: elif normalized_score <= -45:
result.composite_signal = SignalStrength.STRONG_SELL result.composite_signal = SignalStrength.STRONG_SELL
elif normalized_score <= -15: elif normalized_score <= -20:
result.composite_signal = SignalStrength.SELL result.composite_signal = SignalStrength.SELL
else: else:
result.composite_signal = SignalStrength.NEUTRAL result.composite_signal = SignalStrength.NEUTRAL
+156 -20
View File
@@ -110,9 +110,13 @@ class SignalAggregator:
timeframes.append(bt.regime_filter_timeframe) timeframes.append(bt.regime_filter_timeframe)
dfs: Dict[Timeframe, pd.DataFrame] = {} dfs: Dict[Timeframe, pd.DataFrame] = {}
stale_tfs = []
for tf in timeframes: for tf in timeframes:
try: try:
dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf) df = self.data_provider.fetch_ohlcv(symbol, tf)
dfs[tf] = df
if not self.data_provider._is_data_fresh(df, symbol, tf):
stale_tfs.append(tf.value)
except Exception as e: except Exception as e:
logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}") logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
@@ -123,6 +127,14 @@ class SignalAggregator:
use_news=True, use_news=True,
) )
if stale_tfs:
report.warnings.append(f"STALE DATA on {', '.join(stale_tfs)} — signal may be unreliable")
if self.config.primary_timeframe.value in stale_tfs:
report.final_signal = SignalStrength.NEUTRAL
report.signal_quality = "F"
report.trade_setup = None
report.warnings.append("Primary timeframe stale — signal suppressed")
if bt.regime_filter: if bt.regime_filter:
self._apply_regime_filter_live(report, dfs) self._apply_regime_filter_live(report, dfs)
@@ -206,12 +218,12 @@ class SignalAggregator:
minus_dm = -low.diff() minus_dm = -low.diff()
plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0) 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) 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() atr_s = tr.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s)
minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s) minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s)
denom = (plus_di + minus_di).replace(0, float("nan")) denom = (plus_di + minus_di).replace(0, float("nan"))
dx = 100 * (plus_di - minus_di).abs() / denom dx = 100 * (plus_di - minus_di).abs() / denom
adx = dx.ewm(alpha=1/period, min_periods=period).mean().dropna() adx = dx.ewm(alpha=1/period, min_periods=period, adjust=False).mean().dropna()
if adx.empty: if adx.empty:
return None return None
return float(adx.iloc[-1]) return float(adx.iloc[-1])
@@ -308,9 +320,32 @@ class SignalAggregator:
# ── 6. Final signal ──────────────────────────────────────── # ── 6. Final signal ────────────────────────────────────────
self._compute_final_signal(report) self._compute_final_signal(report)
# ── 6b. Session filter (works for both live and backtest) ─
if self.config.backtest.session_filter:
try:
bar_hour = pd.Timestamp(report.timestamp).hour
if bar_hour in self.config.backtest.blocked_hours_utc:
report.final_signal = SignalStrength.NEUTRAL
report.signal_quality = "F"
report.warnings.append(f"SESSION FILTER: {bar_hour:02}:00 UTC blocked")
except Exception:
pass
# ── 7. Trade setup for actionable signals ────────────────── # ── 7. Trade setup for actionable signals ──────────────────
if report.signal_quality in ("A+", "A", "B", "C"): if report.signal_quality in ("A+", "A", "B", "C"):
report.trade_setup = self._generate_trade_setup(report, pip_value) is_sell_direction = (
report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL)
or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0)
)
if is_sell_direction and self.config.backtest.sell_filter:
if report.display_name in self.config.backtest.sell_allowed_pairs:
report.trade_setup = self._generate_trade_setup(report, pip_value)
elif self._forex_sell_qualifies(report):
report.trade_setup = self._generate_trade_setup(report, pip_value)
else:
report.trade_setup = None
else:
report.trade_setup = self._generate_trade_setup(report, pip_value)
# ── 8. Reasons ───────────────────────────────────────────── # ── 8. Reasons ─────────────────────────────────────────────
self._collect_reasons(report) self._collect_reasons(report)
@@ -319,6 +354,11 @@ class SignalAggregator:
# ── Internals ────────────────────────────────────────────────── # ── Internals ──────────────────────────────────────────────────
def _forex_sell_qualifies(self, report: SignalReport) -> bool:
"""Always returns False — forex SELL is structurally unprofitable.
0/4 trades won even under best conditions (trending_down + MTF + ADX>=25 + RSI<=50)."""
return False
def _check_mtf_agreement(self, report: SignalReport) -> bool: def _check_mtf_agreement(self, report: SignalReport) -> bool:
signals = list(report.mtf_signals.values()) signals = list(report.mtf_signals.values())
if len(signals) < 2: if len(signals) < 2:
@@ -398,16 +438,90 @@ class SignalAggregator:
) )
confidence -= 10 confidence -= 10
atr = primary.key_levels.get("atr", 0)
if atr > 0 and score != 0:
is_buy_signal = score > 0
if is_buy_signal:
swing_lows = sorted(
v for k, v in primary.key_levels.items()
if k.startswith("swing_low_")
)
if swing_lows and report.current_price - swing_lows[0] > 2.5 * atr:
confidence -= 15
report.warnings.append("TREND EXHAUSTION — price extended >2.5 ATR past swing low")
else:
swing_highs = sorted(
(v for k, v in primary.key_levels.items()
if k.startswith("swing_high_")),
reverse=True,
)
if swing_highs and swing_highs[0] - report.current_price > 2.5 * atr:
confidence -= 15
report.warnings.append("TREND EXHAUSTION — price extended >2.5 ATR past swing high")
ema_medium = primary.key_levels.get("ema_medium")
if atr > 0 and ema_medium and report.market_regime.startswith("trending"):
pullback_dist = abs(report.current_price - ema_medium) / atr
if pullback_dist > 2.0:
confidence -= 20
report.warnings.append(
f"NO PULLBACK — price {pullback_dist:.1f} ATR from EMA21, chasing risk"
)
elif pullback_dist > 1.5:
confidence -= 10
report.warnings.append(
f"Weak pullback — price {pullback_dist:.1f} ATR from EMA21"
)
if report.market_regime.startswith("trending"):
rsi_ind = primary.indicators.get("rsi")
bb_ind = primary.indicators.get("bollinger")
if rsi_ind:
rsi_val = rsi_ind.value
if report.market_regime in ("trending_up", "trending") and score > 0:
if rsi_val > 65:
confidence -= 15
report.warnings.append(
f"Extended entry — RSI {rsi_val:.0f} in uptrend (prefer pullback to 40-55)"
)
elif 38 <= rsi_val <= 55:
confidence += 5
elif report.market_regime == "trending_down" and score < 0:
if rsi_val < 35:
confidence -= 15
report.warnings.append(
f"Extended entry — RSI {rsi_val:.0f} in downtrend (prefer bounce to 45-62)"
)
elif 45 <= rsi_val <= 62:
confidence += 5
if bb_ind:
bb_pos = bb_ind.value
if report.market_regime in ("trending_up", "trending") and score > 0:
if bb_pos > 0.85:
confidence -= 10
elif bb_pos < 0.4:
confidence += 5
elif report.market_regime == "trending_down" and score < 0:
if bb_pos < 0.15:
confidence -= 10
elif bb_pos > 0.6:
confidence += 5
confidence = max(0.0, min(95.0, confidence)) confidence = max(0.0, min(95.0, confidence))
# Final signal # Final signal
if score >= 35 and confidence >= 60: if score > 38:
confidence -= 20
report.warnings.append("EXHAUSTION: score > 38 — trend overextended")
if score >= 40 and confidence >= 60:
report.final_signal = SignalStrength.STRONG_BUY report.final_signal = SignalStrength.STRONG_BUY
elif score >= 12 and confidence >= 45: elif score >= 30 and confidence >= 55:
report.final_signal = SignalStrength.BUY report.final_signal = SignalStrength.BUY
elif score <= -35 and confidence >= 60: elif score <= -40 and confidence >= 60:
report.final_signal = SignalStrength.STRONG_SELL report.final_signal = SignalStrength.STRONG_SELL
elif score <= -12 and confidence >= 45: elif score <= -30 and confidence >= 55:
report.final_signal = SignalStrength.SELL report.final_signal = SignalStrength.SELL
else: else:
report.final_signal = SignalStrength.NEUTRAL report.final_signal = SignalStrength.NEUTRAL
@@ -415,17 +529,29 @@ class SignalAggregator:
report.final_score = score report.final_score = score
report.final_confidence = confidence report.final_confidence = confidence
# SELL suppression (pair-aware)
sell_suppressed = False
if self.config.backtest.sell_filter and report.final_signal in (
SignalStrength.SELL, SignalStrength.STRONG_SELL
):
if report.display_name not in self.config.backtest.sell_allowed_pairs:
report.final_signal = SignalStrength.NEUTRAL
report.warnings.append(f"SELL SUPPRESSED for {report.display_name}")
sell_suppressed = True
# Quality grade # Quality grade
abs_score = abs(score) abs_score = abs(score)
if abs_score >= 45 and confidence >= 70 and report.mtf_agreement: if sell_suppressed:
report.signal_quality = "F"
elif abs_score >= 45 and confidence >= 70 and report.mtf_agreement:
report.signal_quality = "A+" report.signal_quality = "A+"
elif abs_score >= 30 and confidence >= 60 and report.mtf_agreement: elif abs_score >= 38 and confidence >= 60 and report.mtf_agreement:
report.signal_quality = "A" report.signal_quality = "A"
elif abs_score >= 20 and confidence >= 50: elif abs_score >= 33 and confidence >= 55:
report.signal_quality = "B" report.signal_quality = "B"
elif abs_score >= 12 and confidence >= 40: elif abs_score >= 30 and confidence >= 50:
report.signal_quality = "C" report.signal_quality = "C"
elif abs_score >= 8: elif abs_score >= 15:
report.signal_quality = "D" report.signal_quality = "D"
else: else:
report.signal_quality = "F" report.signal_quality = "F"
@@ -450,7 +576,15 @@ class SignalAggregator:
if atr == 0: if atr == 0:
return None return None
is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY) min_sl_pips = 8.0 if pip_value < 0.01 else 80.0
min_sl_price = min_sl_pips * pip_value
if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
is_buy = True
elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
is_buy = False
else:
is_buy = report.final_score >= 0
levels = report.primary_analysis.key_levels levels = report.primary_analysis.key_levels
min_rr = self.config.signals.min_rr_ratio min_rr = self.config.signals.min_rr_ratio
@@ -467,8 +601,9 @@ class SignalAggregator:
sl_price = swing_lows[0] - atr * 1.0 sl_price = swing_lows[0] - atr * 1.0
else: else:
sl_price = price - atr * 2.0 sl_price = price - atr * 2.0
if price - sl_price < atr * 1.5: min_sl = max(atr * 1.5, min_sl_price)
sl_price = price - atr * 1.5 if price - sl_price < min_sl:
sl_price = price - min_sl
sl_distance = price - sl_price sl_distance = price - sl_price
@@ -496,8 +631,9 @@ class SignalAggregator:
sl_price = swing_highs[0] + atr * 1.0 sl_price = swing_highs[0] + atr * 1.0
else: else:
sl_price = price + atr * 2.0 sl_price = price + atr * 2.0
if sl_price - price < atr * 1.5: min_sl = max(atr * 1.5, min_sl_price)
sl_price = price + atr * 1.5 if sl_price - price < min_sl:
sl_price = price + min_sl
sl_distance = sl_price - price sl_distance = sl_price - price
+60 -20
View File
@@ -54,7 +54,7 @@ from telegram.ext import (
from config import ( from config import (
PlatformConfig, SignalStrength, Timeframe, PlatformConfig, SignalStrength, Timeframe,
EURUSD, USDJPY, GBPUSD, XAUUSD, PAIR_INFO, EURUSD, GBPUSD, XAUUSD, PAIR_INFO,
) )
from data_provider import ForexDataProvider from data_provider import ForexDataProvider
from signal_aggregator import SignalAggregator, SignalReport from signal_aggregator import SignalAggregator, SignalReport
@@ -77,11 +77,10 @@ CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6025899022")
# ── Pairs & buttons ─────────────────────────────────────────────────── # ── Pairs & buttons ───────────────────────────────────────────────────
SCAN_PAIRS = [EURUSD, USDJPY, GBPUSD, XAUUSD] SCAN_PAIRS = [EURUSD, GBPUSD, XAUUSD]
CORRELATION_GROUPS = { CORRELATION_GROUPS = {
"usd_negative": {"pairs": {EURUSD, GBPUSD}, "label": "USD-weakness"}, "usd_negative": {"pairs": {EURUSD, GBPUSD}, "label": "USD-weakness"},
"risk_sentiment": {"pairs": {XAUUSD, USDJPY}, "label": "Risk-sentiment"},
} }
_WS_TO_PAIR = { _WS_TO_PAIR = {
@@ -93,17 +92,14 @@ _WS_SYMBOLS = list(_WS_TO_PAIR.keys())
_LEVEL_KEYS = ( _LEVEL_KEYS = (
"swing_high_1", "swing_high_2", "swing_high_3", "swing_high_1", "swing_high_2", "swing_high_3",
"swing_low_1", "swing_low_2", "swing_low_3", "swing_low_1", "swing_low_2", "swing_low_3",
"bb_upper", "bb_lower", "ema_trend", "ema_trend",
) )
SCAN_BUTTONS = [ SCAN_BUTTONS = [
{"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"}, {"cb": "scan_eurusd_m15", "symbol": EURUSD, "tf": Timeframe.M15, "label": "EUR/USD 15M"},
{"cb": "scan_eurusd_h1", "symbol": EURUSD, "tf": Timeframe.H1, "label": "EUR/USD 1H"}, {"cb": "scan_eurusd_h1", "symbol": EURUSD, "tf": Timeframe.H1, "label": "EUR/USD 1H"},
{"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"}, {"cb": "scan_eurusd_h4", "symbol": EURUSD, "tf": Timeframe.H4, "label": "EUR/USD 4H"},
{"cb": "scan_usdjpy_m15", "symbol": USDJPY, "tf": Timeframe.M15, "label": "USD/JPY 15M"}, {"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
{"cb": "scan_usdjpy_h1", "symbol": USDJPY, "tf": Timeframe.H1, "label": "USD/JPY 1H"},
{"cb": "scan_usdjpy_h4", "symbol": USDJPY, "tf": Timeframe.H4, "label": "USD/JPY 4H"},
{"cb": "scan_gbpusd_m15", "symbol": GBPUSD, "tf": Timeframe.M15, "label": "GBP/USD 15M"},
{"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"}, {"cb": "scan_gbpusd_h1", "symbol": GBPUSD, "tf": Timeframe.H1, "label": "GBP/USD 1H"},
{"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"}, {"cb": "scan_gbpusd_h4", "symbol": GBPUSD, "tf": Timeframe.H4, "label": "GBP/USD 4H"},
{"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"}, {"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU/USD 15M"},
@@ -156,12 +152,12 @@ _QUALITY_EMOJI = {
"A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌", "A+": "🏆", "A": "🥇", "B": "🥈", "C": "🥉", "D": "⚠️", "F": "❌",
} }
_ARROW = { _SIGNAL_TAG = {
SignalStrength.STRONG_BUY: "⬆️", SignalStrength.STRONG_BUY: "Strong Buy",
SignalStrength.BUY: "↗️", SignalStrength.BUY: "Bullish",
SignalStrength.NEUTRAL: "➡️", SignalStrength.NEUTRAL: "Neutral",
SignalStrength.SELL: "↘️", SignalStrength.SELL: "Bearish",
SignalStrength.STRONG_SELL: "⬇️", SignalStrength.STRONG_SELL: "Strong Sell",
} }
@@ -180,9 +176,9 @@ def _mtf_line(report: SignalReport) -> str:
def _indicator_line(name: str, ind) -> str: def _indicator_line(name: str, ind) -> str:
arrow = _ARROW.get(ind.signal, "➡️") tag = _SIGNAL_TAG.get(ind.signal, "Neutral")
short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description short = ind.description.split(" | ")[0] if " | " in ind.description else ind.description
return f"{arrow} *{ind.name}*: {short}" return f"[{tag}] *{ind.name}*: {short}"
def format_signal_message(report: SignalReport) -> str: def format_signal_message(report: SignalReport) -> str:
@@ -601,9 +597,11 @@ async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
text = ( text = (
"📖 *ForexBot2 Commands*\n\n" "📖 *ForexBot2 Commands*\n\n"
"/start — Show scan buttons (4 pairs × 4 TFs)\n" "/start — Show scan buttons\n"
"/autoscan — Toggle auto-scan on/off\n" "/autoscan — Toggle auto-scan on/off\n"
"/status — Bot status, WS connection, live prices\n" "/status — Bot status, WS connection, live prices\n"
"/cache — Show when each pair/TF was last fetched\n"
"/clearcache — Wipe data cache, force fresh fetch\n"
"/help — This message\n\n" "/help — This message\n\n"
"*Auto-scan* fires at candle close: M5/M15/H1/H4.\n" "*Auto-scan* fires at candle close: M5/M15/H1/H4.\n"
"*WebSocket* monitors live prices and sends Level Break Alerts " "*WebSocket* monitors live prices and sends Level Break Alerts "
@@ -628,10 +626,10 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
lines.append("❌ WebSocket: *disconnected*") lines.append("❌ WebSocket: *disconnected*")
if _last_prices: if _last_prices:
lines.append("\n*Live Prices:*") lines.append("\n*Live Prices (WebSocket):*")
for pair in SCAN_PAIRS: for ws_sym, pair in _WS_TO_PAIR.items():
price = _last_prices.get(pair) price = _last_prices.get(pair)
display = PAIR_INFO.get(pair, {}).get("display", pair) display = PAIR_INFO.get(pair, {}).get("display", ws_sym)
if price is None: if price is None:
lines.append(f" {display}: waiting...") lines.append(f" {display}: waiting...")
elif price < 10: elif price < 10:
@@ -658,6 +656,44 @@ async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN) await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
async def cmd_cache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
dp = _get_shared_dp()
if not dp._cache_timestamps:
await update.message.reply_text("📭 Cache is empty — no data fetched yet.")
return
now = time.time()
lines = ["📦 *Data Cache Status*\n"]
for pair in SCAN_PAIRS:
display = PAIR_INFO.get(pair, {}).get("display", pair)
pair_lines = []
for tf in [Timeframe.M5, Timeframe.M15, Timeframe.H1, Timeframe.H4, Timeframe.D1]:
cache_key = f"{pair}_{tf.value}"
ts = dp._cache_timestamps.get(cache_key)
if ts is None:
continue
age_min = (now - ts) / 60
df = dp._cache.get(cache_key)
last_bar = df.index[-1].strftime("%H:%M") if df is not None and not df.empty else "?"
fresh = "✅" if age_min < dp.CANDLE_SECONDS.get(tf, 3600) / 60 else "⚠️"
pair_lines.append(f" {tf.value:<5} {fresh} fetched `{age_min:.0f}m` ago | last bar `{last_bar}`")
if pair_lines:
lines.append(f"*{display}*")
lines.extend(pair_lines)
await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN)
async def cmd_clearcache(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
dp = _get_shared_dp()
count = dp.clear_cache()
await update.message.reply_text(
f"🗑 Cache cleared ({count} entries wiped).\nNext scan will fetch fresh data from API.",
parse_mode=ParseMode.MARKDOWN,
)
async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: async def cmd_autoscan(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
chat_id = update.effective_chat.id chat_id = update.effective_chat.id
@@ -725,6 +761,8 @@ async def _post_init(app: Application) -> None:
("start", "Show scan buttons"), ("start", "Show scan buttons"),
("autoscan", "Toggle auto-scan on/off"), ("autoscan", "Toggle auto-scan on/off"),
("status", "Bot status, WS connection, live prices"), ("status", "Bot status, WS connection, live prices"),
("cache", "Show data cache age per pair/TF"),
("clearcache", "Wipe data cache, force fresh fetch"),
("help", "All commands"), ("help", "All commands"),
]) ])
@@ -792,6 +830,8 @@ def main() -> None:
app.add_handler(CommandHandler("help", cmd_help)) app.add_handler(CommandHandler("help", cmd_help))
app.add_handler(CommandHandler("status", cmd_status)) app.add_handler(CommandHandler("status", cmd_status))
app.add_handler(CommandHandler("autoscan", cmd_autoscan)) app.add_handler(CommandHandler("autoscan", cmd_autoscan))
app.add_handler(CommandHandler("cache", cmd_cache))
app.add_handler(CommandHandler("clearcache", cmd_clearcache))
app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_")) app.add_handler(CallbackQueryHandler(cb_scan, pattern="^scan_"))
logger.info("Bot running — press Ctrl+C to stop") logger.info("Bot running — press Ctrl+C to stop")