harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Narrow bot to USDCAD H4 Grade B + paper-trade tracker + /status command

Browse Source 
Backtest-driven config change. The previous wide-scan setup (4 pairs ×
M15+H1, alert grades A+/A) was empirically losing money in every
backtested permutation. The 2-year backtest identified one robust niche
(USDCAD H4 Grade B: train +20.8 R / test +14.0 R / PF 1.52 / max DD
-9.6 R, positive in every 6-month sub-period). This commit restricts
the live bot to that niche and runs it in paper-trade mode for
validation before any real capital is risked.

Config:
  SCAN_PAIRS         ["USDCAD=X"]   (was 4 pairs)
  SCAN_TIMEFRAMES    ["H4"]          (was M15+H1)
  ALERT_GRADES       {"B"}           (was A+/A — those are anti-edge)
  AUTO_SCAN_INTERVAL 1800            (was 600 — H4 candles only print
                                       every 4h, polling more wastes credits)
  PAPER_TRADE_MODE   True            (alerts get [PAPER] prefix and
                                       are NOT for real money)

New machinery:
- _open_paper_position() / _check_paper_positions(): on every scan,
  walks any open paper trades forward against the latest H4 bars using
  the same first-touch SL/TP1 logic as backtest.simulate_trade.
- paper_open.json + paper_trades.csv: persistent state across restarts.
  Open positions survive bot restarts; closed trades append to the CSV.
- /status: uptime, scan count, last scan time, alert count, open paper
  positions, win rate, total R, profit factor.
- /paper: full ledger — open trades + last 10 closed.
- INFO-level logging in main() so the user sees scan activity in
  stdout instead of wondering whether the bot is alive.
- Help text updated to reflect the calibrated strategy and warn that
  A+/A grades are anti-edge in backtest.
- gitignore: paper_open.json + paper_trades.csv + backtest_results_*/

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Harwinder
2026-05-06 15:01:35 -04:00
parent 988adf9a3d
commit ced4efa96d
2 changed files with 440 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+5
View File
@@ -19,3 +19,8 @@ Thumbs.db
# Backtest data + outputs (regenerable from seed_history.py / backtest.py)
history/
backtest_results/
backtest_results_*/
# Paper-trade ledger (live state — do not version control)
paper_open.json
paper_trades.csv
telegram_bot.py
+435 -34
View File
@@ -17,14 +17,19 @@ Commands:
"""
import asyncio
import csv
import html
import json
import logging
import os
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from zoneinfo import ZoneInfo
import pandas as pd
# Load .env so TELEGRAM_BOT_TOKEN / GEMINI_API_KEY / TWELVE_DATA_API_KEY
# don't need to be exported manually.
try:
@@ -48,17 +53,32 @@ logger = logging.getLogger(__name__)
# ── Config ──────────────────────────────────────────────────────────
SCAN_TIMEFRAMES = ["M15", "H1"] # two cards (M15 and H1) per /scan; cache shared
# ─────────────────────────────────────────────────────────────────────
# Strategy config — calibrated by 2-year backtest (May 2024 → May 2026)
# ─────────────────────────────────────────────────────────────────────
# Backtest verdict: M15 has zero edge across 9,659 trades. Only USDCAD
# on H4 at Grade B survived 2-year regime stability — train +20.8 R,
# test +14.0 R, PF 1.52, max DD -9.6 R, positive in every 6-month
# sub-period. Grades A+/A are anti-edge (highest grade ≠ best trades).
# See backtest_results_h4_2y/ for the data.
# Pairs analyzed by the /scan command. Limited to the most liquid majors
# to keep credits low and signal quality high.
SCAN_PAIRS = ["EURUSD=X", "USDJPY=X", "GBPUSD=X", "AUDUSD=X"]
SCAN_TIMEFRAMES = ["H4"] # H4 only — M15 has no edge
SCAN_PAIRS = ["USDCAD=X"] # USDCAD only — only pair with stable edge
ALERT_GRADES = {"B"} # B only — A+/A are anti-edge
# Auto-scan job: fires every AUTO_SCAN_INTERVAL seconds and pushes any A/A+
# signal to ALERT_CHAT_ID (env var). 10 min cadence ≈ 595 credits/day on
# cold-cache average — fits inside the 800/day Twelve Data free-tier budget.
AUTO_SCAN_INTERVAL = 600
ALERT_GRADES = {"A+", "A"}
# 30-min cadence: H4 candles update only every 4 hours, so polling more
# often wastes API credits. 48 scans/day × ~5 credits each ≈ 240/day,
# well inside the 800/day free-tier budget.
AUTO_SCAN_INTERVAL = 1800
# Paper-trade mode: when True, alerts get a [PAPER] prefix and are
# logged to paper_trades.csv so we can validate the strategy against
# the backtest distribution before risking real money.
PAPER_TRADE_MODE = True
# Where paper-trade state lives (gitignored)
_PAPER_OPEN_FILE = Path("paper_open.json")
_PAPER_TRADES_FILE = Path("paper_trades.csv")
_TF_MAP = {
"M5": Timeframe.M5, "M15": Timeframe.M15,
@@ -66,11 +86,11 @@ _TF_MAP = {
"H4": Timeframe.H4, "D1": Timeframe.D1,
}
# D1 kept as confirmation for higher-timeframe trend context.
# W1 dropped — only the D1 scan used it, and we no longer scan D1.
# Confirmation TFs per primary — mirrors backtest.py's mapping
_TF_CONFIRMATIONS = {
"M15": [Timeframe.H1, Timeframe.D1],
"H1": [Timeframe.M15, Timeframe.D1],
"H4": [Timeframe.H1, Timeframe.D1],
}
_GRADE_RANK = {"A+": 0, "A": 1, "B": 2, "C": 3, "D": 4, "F": 5}
@@ -281,11 +301,17 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
sym = _grade_symbol(r.signal_quality)
action = _signal_action(r.final_signal)
paper_tag = "📝 [PAPER] " if PAPER_TRADE_MODE else ""
lines.append(
f"{sym} <b>{_h(r.signal_quality)} ALERT · {_h(r.display_name)}"
f"{paper_tag}{sym} <b>{_h(r.signal_quality)} ALERT · {_h(r.display_name)}"
f" · {action} · {tf}</b>"
)
lines.append(f"🕐 {now_str}")
if PAPER_TRADE_MODE:
lines.append(
"<i>Paper-trade mode — this is logged to paper_trades.csv. "
"Do NOT trade with real money until validation period ends.</i>"
)
lines.append("")
# ── Trade setup ──────────────────────────────────────────────
@@ -374,6 +400,52 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
return "\n".join(lines)
def _walk_forward(pos: dict, forward_bars: pd.DataFrame, spread_pips: float) -> tuple:
"""
First-touch SL/TP1 simulation on the primary timeframe — same logic as
backtest.simulate_trade. Returns (outcome, exit_price, exit_ts, bars_held, r_multiple).
Outcome is OPEN, WIN_TP1, LOSS_SL, or TIMEOUT.
"""
if pos["risk_pips"] <= 0:
return "INVALID", 0.0, None, 0, 0.0
spread_r = spread_pips / pos["risk_pips"]
bars = 0
for ts, bar in forward_bars.iterrows():
bars += 1
if bars > pos["max_hold"]:
break
if pos["direction"] == "BUY":
sl_hit = bar["Low"] <= pos["sl"]
tp_hit = bar["High"] >= pos["tp1"]
else:
sl_hit = bar["High"] >= pos["sl"]
tp_hit = bar["Low"] <= pos["tp1"]
if sl_hit:
return "LOSS_SL", float(pos["sl"]), ts, bars, -1.0 - spread_r
if tp_hit:
r = (pos["reward_pips"] / pos["risk_pips"]) - spread_r
return "WIN_TP1", float(pos["tp1"]), ts, bars, r
if bars == 0:
return "OPEN", 0.0, None, 0, 0.0
if bars >= pos["max_hold"]:
last_idx = bars - 1
last = forward_bars.iloc[last_idx]
last_ts = forward_bars.index[last_idx]
if pos["direction"] == "BUY":
pnl_price = last["Close"] - pos["entry_price"]
else:
pnl_price = pos["entry_price"] - last["Close"]
sl_dist = abs(pos["entry_price"] - pos["sl"])
r = (pnl_price / sl_dist) - spread_r if sl_dist > 0 else 0.0
return "TIMEOUT", float(last["Close"]), last_ts, bars, r
# Reached end of available bars without SL/TP/timeout — still open
return "OPEN", 0.0, None, bars, 0.0
# ══════════════════════════════════════════════════════════════════════
# Bot
# ══════════════════════════════════════════════════════════════════════
@@ -381,7 +453,7 @@ def _build_alert_card(tf: str, r: SignalReport, now_str: str) -> str:
class ForexTelegramBot:
def __init__(self):
self._executor = ThreadPoolExecutor(
max_workers=len(SCAN_TIMEFRAMES), thread_name_prefix="forex"
max_workers=max(1, len(SCAN_TIMEFRAMES)), thread_name_prefix="forex"
)
# Persistent ForexDataProvider so the OHLCV cache survives across
# /scan and /signal calls. Without this each call burns API credits
@@ -393,6 +465,17 @@ class ForexTelegramBot:
# setup persists. Bounded to avoid unbounded growth on long uptime.
self._alerted_keys: set = set()
# Auto-scan visibility — exposed by /status
self._scan_count = 0
self._alert_count = 0
self._last_scan_at: Optional[datetime] = None
self._last_alert_at: Optional[datetime] = None
self._started_at = datetime.now(timezone.utc)
# Paper-trade tracker: open positions persist across restarts via
# paper_open.json; closed trades append to paper_trades.csv.
self._paper_open: List[dict] = self._load_paper_open()
# ── Engine helpers ────────────────────────────────────────────────
def _build_config(self, tf: str, pairs: List[str]) -> PlatformConfig:
@@ -461,6 +544,133 @@ class ForexTelegramBot:
async def cmd_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
await self.cmd_help(update, context)
async def cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
/status — proves the auto-scan is alive and shows a summary.
Reports: uptime, scan count, last scan time, alert count, open
paper positions, paper-trade P&L (R-units).
"""
now = datetime.now(timezone.utc)
uptime = now - self._started_at
h, rem = divmod(int(uptime.total_seconds()), 3600)
m, _ = divmod(rem, 60)
last_scan = (
f"{(now - self._last_scan_at).total_seconds()/60:.1f} min ago "
f"({self._last_scan_at.strftime('%H:%M:%S UTC')})"
if self._last_scan_at else "never (waiting for first scan)"
)
last_alert = (
f"{(now - self._last_alert_at).total_seconds()/60:.1f} min ago"
if self._last_alert_at else "no alerts yet"
)
# Paper P&L summary
paper_summary = self._paper_summary_lines()
text = (
f"🤖 <b>Bot Status</b>\n\n"
f"<b>Strategy:</b> {SCAN_PAIRS} on {SCAN_TIMEFRAMES} · "
f"grades {sorted(ALERT_GRADES)}"
f"{' · 📝 PAPER MODE' if PAPER_TRADE_MODE else ''}\n"
f"<b>Cadence:</b> every {AUTO_SCAN_INTERVAL//60} min\n\n"
f"<b>Uptime:</b> {h}h {m}m\n"
f"<b>Scans run:</b> {self._scan_count}\n"
f"<b>Last scan:</b> {last_scan}\n"
f"<b>Total alerts:</b> {self._alert_count}\n"
f"<b>Last alert:</b> {last_alert}\n\n"
f"{paper_summary}"
)
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
async def cmd_paper(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
/paper — full paper-trade ledger summary: open positions and last 10 closed.
"""
text = self._paper_full_report()
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
def _paper_summary_lines(self) -> str:
"""Short paper-trade stats block for /status."""
n_open = len(self._paper_open)
if not _PAPER_TRADES_FILE.exists():
return (
f"<b>Paper trades:</b>\n"
f" Open: {n_open}\n"
f" Closed: 0 (no trades closed yet)"
)
try:
df = pd.read_csv(_PAPER_TRADES_FILE)
except Exception:
return f"<b>Paper trades:</b>\n Open: {n_open} (csv unreadable)"
if df.empty:
return f"<b>Paper trades:</b>\n Open: {n_open}\n Closed: 0"
n = len(df)
wr = (df["r_multiple"] > 0).mean() * 100
total_r = df["r_multiple"].sum()
avg_r = df["r_multiple"].mean()
return (
f"<b>Paper trades:</b>\n"
f" Open: {n_open}\n"
f" Closed: {n} · WR {wr:.1f}% · total R <b>{total_r:+.2f}</b> · avg R {avg_r:+.3f}\n"
f"<i>(backtest expectation: ~+0.30 R/trade for USDCAD H4 B)</i>"
)
def _paper_full_report(self) -> str:
lines = ["📝 <b>Paper-trade ledger</b>", ""]
# Open
if self._paper_open:
lines.append(f"<b>Open ({len(self._paper_open)}):</b>")
for p in self._paper_open:
lines.append(
f" • {_h(p['pair'])} {p['tf']} {p['direction']} "
f"@ {p['entry_price']:.5f} "
f"SL {p['sl']:.5f} / TP {p['tp1']:.5f} "
f"({p.get('bars_held', 0)}/{p['max_hold']} bars)"
)
else:
lines.append("<b>Open:</b> none")
lines.append("")
# Closed (last 10)
if not _PAPER_TRADES_FILE.exists():
lines.append("<b>Closed:</b> none yet")
return "\n".join(lines)
try:
df = pd.read_csv(_PAPER_TRADES_FILE)
except Exception as e:
lines.append(f"<b>Closed:</b> error reading {_PAPER_TRADES_FILE}: {e}")
return "\n".join(lines)
if df.empty:
lines.append("<b>Closed:</b> none yet")
return "\n".join(lines)
n = len(df)
wr = (df["r_multiple"] > 0).mean() * 100
total_r = df["r_multiple"].sum()
avg_r = df["r_multiple"].mean()
gross_win = df.loc[df["r_multiple"] > 0, "r_multiple"].sum()
gross_loss = -df.loc[df["r_multiple"] <= 0, "r_multiple"].sum()
pf = gross_win / gross_loss if gross_loss > 0 else float("inf")
lines.append(
f"<b>Closed ({n}):</b> WR {wr:.1f}% · totalR <b>{total_r:+.2f}</b> · "
f"avgR {avg_r:+.3f} · PF {pf:.2f}"
)
lines.append("")
lines.append("<b>Last 10 closed:</b>")
recent = df.tail(10)
for _, row in recent.iterrows():
emoji = "✅" if row["r_multiple"] > 0 else "❌"
lines.append(
f" {emoji} {_h(row['pair'])} {row['tf']} {row['direction']} "
f"{row['outcome']} R={row['r_multiple']:+.2f}"
)
return "\n".join(lines)
async def cmd_myid(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""
Returns the current chat's ID so the operator can set ALERT_CHAT_ID
@@ -475,29 +685,185 @@ class ForexTelegramBot:
)
await update.message.reply_text(text, parse_mode=ParseMode.HTML)
# ── Paper-trade tracker ──────────────────────────────────────────
@staticmethod
def _load_paper_open() -> List[dict]:
if not _PAPER_OPEN_FILE.exists():
return []
try:
return json.loads(_PAPER_OPEN_FILE.read_text())
except Exception as e:
logger.warning(f"Could not load {_PAPER_OPEN_FILE}: {e}")
return []
def _save_paper_open(self) -> None:
try:
_PAPER_OPEN_FILE.write_text(json.dumps(self._paper_open, indent=2))
except Exception as e:
logger.error(f"Could not save {_PAPER_OPEN_FILE}: {e}")
def _append_closed_trade(self, row: dict) -> None:
"""Append one closed-trade row to paper_trades.csv (header on first write)."""
write_header = not _PAPER_TRADES_FILE.exists()
try:
with _PAPER_TRADES_FILE.open("a", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(row.keys()))
if write_header:
w.writeheader()
w.writerow(row)
except Exception as e:
logger.error(f"Could not append to {_PAPER_TRADES_FILE}: {e}")
def _open_paper_position(self, tf: str, r: SignalReport) -> None:
"""Record a fresh alert as an open paper position."""
if r.trade_setup is None:
return
ts = r.trade_setup
bar_ts = r.primary_analysis.timestamp if r.primary_analysis else ""
pos = {
"alert_id": f"{r.symbol}|{tf}|{ts.direction}|{bar_ts}",
"symbol": r.symbol,
"pair": r.display_name,
"tf": tf,
"grade": r.signal_quality,
"direction": ts.direction,
"entry_time": bar_ts, # bar's close timestamp
"entry_price": ts.entry_price,
"sl": ts.stop_loss,
"tp1": ts.take_profit_1,
"tp2": ts.take_profit_2,
"risk_pips": ts.risk_pips,
"reward_pips": ts.reward_pips,
"score": r.final_score,
"confidence": r.final_confidence,
"regime": r.market_regime,
"mtf_aligned": bool(r.mtf_agreement),
"max_hold": 100,
"status": "OPEN",
}
# Don't double-open if alert_id already in flight
if any(p["alert_id"] == pos["alert_id"] for p in self._paper_open):
return
self._paper_open.append(pos)
self._save_paper_open()
logger.info(f"📝 paper_open: {pos['pair']} {tf} {ts.direction} @ {ts.entry_price:.5f}")
async def _check_paper_positions(self, context: ContextTypes.DEFAULT_TYPE) -> None:
"""
Walk forward each open paper position using the warm OHLCV cache.
First-touch SL/TP1 — same logic as backtest.simulate_trade. Closed
positions are appended to paper_trades.csv and a brief notification
is sent to the alert chat.
"""
if not self._paper_open:
return
chat_id = context.job.data["chat_id"]
loop = asyncio.get_running_loop()
still_open: List[dict] = []
spread_pips = 1.0
for pos in self._paper_open:
symbol = pos["symbol"]
tf_enum = _TF_MAP[pos["tf"]]
try:
df = await loop.run_in_executor(
None, self._provider.fetch_ohlcv, symbol, tf_enum,
)
except Exception as e:
logger.warning(f"check_paper: {symbol} fetch failed: {e}")
still_open.append(pos)
continue
entry_ts = pd.Timestamp(pos["entry_time"])
if entry_ts.tz is None:
entry_ts = entry_ts.tz_localize("UTC")
forward = df[df.index > entry_ts].head(pos["max_hold"])
if forward.empty:
still_open.append(pos)
continue
outcome, exit_price, exit_ts, bars_held, r_mult = _walk_forward(
pos, forward, spread_pips,
)
if outcome == "OPEN":
pos["bars_held"] = bars_held
still_open.append(pos)
continue
# Position closed — log to CSV and notify
closed = dict(pos)
closed["status"] = "CLOSED"
closed["outcome"] = outcome
closed["exit_time"] = str(exit_ts)
closed["exit_price"] = exit_price
closed["bars_held"] = bars_held
closed["r_multiple"] = round(r_mult, 4)
self._append_closed_trade(closed)
logger.info(
f"📊 paper_close: {pos['pair']} {pos['tf']} {pos['direction']} "
f"{outcome} R={r_mult:+.2f}"
)
# Notify chat
emoji = "✅" if r_mult > 0 else "❌"
msg = (
f"{emoji} <b>[PAPER] Position closed</b>\n"
f"{_h(pos['pair'])} {pos['tf']} {pos['direction']}\n"
f"Entry {pos['entry_price']:.5f} → Exit {exit_price:.5f}\n"
f"Outcome: <b>{outcome}</b> · R = <b>{r_mult:+.2f}</b>\n"
f"Held {bars_held} {pos['tf']} bars"
)
try:
await context.bot.send_message(
chat_id=chat_id, text=msg, parse_mode=ParseMode.HTML,
)
except Exception as e:
logger.error(f"close-notify failed: {e}")
self._paper_open = still_open
self._save_paper_open()
# ── Auto-scan job (every AUTO_SCAN_INTERVAL seconds) ─────────────
async def _auto_scan_job(self, context: ContextTypes.DEFAULT_TYPE):
"""
Periodic scan that pushes A/A+ alerts to the configured chat.
Dedup: one alert per (symbol, tf, direction, candle_open) — so a
persistent setup re-alerts once per new bar, not every 10 min.
Periodic scan: (1) checks any open paper positions for SL/TP exits,
(2) emits new alerts for setups in ALERT_GRADES, (3) opens fresh
paper positions for every alert.
Dedup: one alert per (symbol, tf, direction, candle_open) — a
persistent setup re-alerts once per new bar, not every scan tick.
"""
chat_id = context.job.data["chat_id"]
# 1. Walk open paper positions forward against the latest bars
await self._check_paper_positions(context)
# 2. Run the engine
try:
results_by_tf = await self._analyze_all_tfs(list(SCAN_PAIRS))
except Exception as e:
logger.error(f"Auto-scan failed: {e}", exc_info=True)
return
# 3. Update visibility counters
self._scan_count += 1
self._last_scan_at = datetime.now(timezone.utc)
logger.info(
f"auto_scan #{self._scan_count}: scanned {SCAN_PAIRS} on "
f"{SCAN_TIMEFRAMES}; open paper positions: {len(self._paper_open)}"
)
# 4. Emit alerts for graded setups
now_str = _now_est()
sent = 0
for tf, reports in results_by_tf.items():
for r in reports:
if r.signal_quality not in ALERT_GRADES:
continue
# Skip if engine blocked it via event danger
if r.event_risk == "danger":
continue
ts = r.primary_analysis.timestamp if r.primary_analysis else ""
@@ -514,35 +880,55 @@ class ForexTelegramBot:
parse_mode=ParseMode.HTML,
)
sent += 1
self._alert_count += 1
self._last_alert_at = datetime.now(timezone.utc)
logger.info(
f"Alert sent: {r.symbol} {tf} {r.final_signal.name} "
f"({r.signal_quality})"
)
# Open a paper position
if PAPER_TRADE_MODE:
self._open_paper_position(tf, r)
except Exception as e:
logger.error(f"Failed to send alert for {r.symbol} {tf}: {e}")
# Bound dedup memory: keep most recent 256 keys
if len(self._alerted_keys) > 256:
self._alerted_keys = set(list(self._alerted_keys)[-256:])
if sent == 0:
logger.debug("Auto-scan complete: no new A-grade signals")
logger.debug(f"auto_scan #{self._scan_count}: no new alerts")
async def cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
tfs = " · ".join(SCAN_TIMEFRAMES)
pairs = " · ".join(p.replace("=X", "") for p in SCAN_PAIRS)
grades = " · ".join(sorted(ALERT_GRADES))
paper_note = (
"📝 <b>Paper-trade mode is ACTIVE.</b> "
"Alerts are tagged [PAPER] and logged to paper_trades.csv. "
"Do NOT use real capital until validation period ends.\n\n"
if PAPER_TRADE_MODE else ""
)
text = (
"🌐 <b>Forex Signal Intelligence Platform</b>\n\n"
f"{paper_note}"
"<b>Strategy (calibrated by 2-yr backtest):</b>\n"
f" Pair: <code>{pairs}</code>\n"
f" Timeframe: <code>{tfs}</code>\n"
f" Alert grades: <code>{grades}</code>\n"
f" Auto-scan: every {AUTO_SCAN_INTERVAL//60} min\n\n"
"<b>Commands:</b>\n"
" /signal EURUSD — Full signal analysis across all timeframes\n"
" /scan — Scan all pairs, compact Grade B+ summary\n"
" /help — This message\n\n"
f"<b>Timeframes analyzed:</b> <code>{tfs}</code>\n\n"
" /status — auto-scan health + paper P&amp;L\n"
" /paper — full paper-trade ledger\n"
" /signal EURUSD — full analysis on one pair (any TF)\n"
" /scan — manual scan of configured pairs\n"
" /myid — chat ID (for ALERT_CHAT_ID setup)\n"
" /help — this message\n\n"
"<b>Signal Grades:</b>\n"
" 🔥 A+ — Strongest signal, MTF aligned, high confidence\n"
" ✅ A — Strong signal, MTF aligned\n"
" 👍 B — Decent setup, wait for confirmation\n"
" ⚠️ C — Weak signal, watch only\n"
" 📉 D — Very weak, no action\n"
" 🔥 A+ — Highest engine score (⚠️ backtest: anti-edge)\n"
" ✅ A — High engine score (⚠️ backtest: anti-edge)\n"
" 👍 B — Mid-confluence (✅ backtest: where the edge lives)\n"
" ⚠️ C — Weak signal\n"
" 📉 D — Very weak\n"
" ❌ F — No clear signal\n\n"
"<i>⚠️ Educational use only. Not financial advice.</i>"
)
@@ -658,8 +1044,10 @@ class ForexTelegramBot:
async def _register_commands(app: Application) -> None:
await app.bot.set_my_commands([
BotCommand("status", "Auto-scan status, scan count, paper P&L"),
BotCommand("paper", "Paper-trade ledger — open positions + last 10 closed"),
BotCommand("signal", "Signal analysis for one pair — /signal EURUSD"),
BotCommand("scan", "Scan all pairs, best Grade B+ per pair"),
BotCommand("scan", "Manual scan of configured pairs"),
BotCommand("myid", "Show this chat's ID (for ALERT_CHAT_ID setup)"),
BotCommand("help", "Commands and grade guide"),
])
@@ -675,6 +1063,8 @@ class ForexTelegramBot:
app.add_handler(CommandHandler("help", self.cmd_help))
app.add_handler(CommandHandler("signal", self.cmd_signal))
app.add_handler(CommandHandler("scan", self.cmd_scan))
app.add_handler(CommandHandler("status", self.cmd_status))
app.add_handler(CommandHandler("paper", self.cmd_paper))
app.add_handler(CommandHandler("myid", self.cmd_myid))
# ── Auto-scan job: only enabled when ALERT_CHAT_ID is set ─────
@@ -713,14 +1103,25 @@ class ForexTelegramBot:
# ══════════════════════════════════════════════════════════════════════
def main():
# INFO level so the user can see auto-scan activity in stdout
# ("auto_scan #5: scanned ['USDCAD=X'] on ['H4'] …").
# Drop external libraries to WARNING to keep the log readable.
logging.basicConfig(
level=logging.WARNING,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
datefmt="%H:%M:%S",
)
for noisy in ("httpx", "httpcore", "telegram", "apscheduler", "asyncio"):
logging.getLogger(noisy).setLevel(logging.WARNING)
bot = ForexTelegramBot()
app = bot.build_application()
print("🤖 Forex Signal Bot started. Press Ctrl+C to stop.")
print("🤖 Forex Signal Bot started.")
print(f" Strategy: {SCAN_PAIRS} on {SCAN_TIMEFRAMES} · grades {sorted(ALERT_GRADES)}")
print(f" Auto-scan: every {AUTO_SCAN_INTERVAL//60} min")
if PAPER_TRADE_MODE:
print(f" 📝 Paper-trade mode ON — logging to {_PAPER_TRADES_FILE}")
print(" Ctrl+C to stop.\n")
app.run_polling(drop_pending_updates=True)