ForexBot2/signal_web.py

#!/usr/bin/env python3
"""
ForexBot2 — Signal Daemon
==========================
Candle-boundary auto-scanner that publishes signal data to Redis.
Mirrors telegram_bot.py scan loops but outputs to Redis instead of Telegram.

Scan loop: M15/H1/H4 at candle boundaries + 10s buffer.
Data flow: SignalAggregator → serialize → Redis key + pub/sub channel.

Usage:
    python signal_web.py
"""

import asyncio
import json
import logging
import os
import signal as unix_signal
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from typing import Dict, List, Optional

import redis

from config import (
    PlatformConfig, SignalStrength, Timeframe,
    EURUSD, GBPUSD, XAUUSD, PAIR_INFO,
)
from data_provider import ForexDataProvider
from signal_aggregator import SignalAggregator, SignalReport

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    level=logging.INFO,
    datefmt="%H:%M:%S",
)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logger = logging.getLogger("signal_web")

# ── Config ────────────────────────────────────────────────────────────

SCAN_PAIRS = [EURUSD, GBPUSD, XAUUSD]

AUTOSCAN_CONFIRMATIONS = {
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
    Timeframe.H4:  [Timeframe.H1, Timeframe.D1],
}

AUTOSCAN_LOOPS = [
    (Timeframe.M15, 15),
    (Timeframe.H1,  60),
    (Timeframe.H4,  240),
]

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
REDIS_KEY = "forexbot2:signals"
REDIS_CHANNEL = "forexbot2:updates"

_executor = ThreadPoolExecutor(max_workers=4)

# ── Shared data provider ──────────────────────────────────────────────

_shared_dp: Optional[ForexDataProvider] = None


def _get_shared_dp() -> ForexDataProvider:
    global _shared_dp
    if _shared_dp is None:
        _shared_dp = ForexDataProvider(PlatformConfig())
    return _shared_dp


# ── Redis client ──────────────────────────────────────────────────────

_redis: Optional[redis.Redis] = None


def _init_redis() -> None:
    global _redis
    _redis = redis.from_url(REDIS_URL, decode_responses=True)
    _redis.ping()
    logger.info(f"Redis connected: {REDIS_URL}")


# ── Signal runner (mirrors telegram_bot.py) ───────────────────────────

TF_CONFIRMATIONS = {
    Timeframe.M15: [Timeframe.M5, Timeframe.H1],
    Timeframe.H1:  [Timeframe.M15, Timeframe.H4],
    Timeframe.H4:  [Timeframe.H1, Timeframe.D1],
}


def _run_signal_sync(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    if confirmation_tfs is None:
        confirmation_tfs = TF_CONFIRMATIONS.get(
            timeframe, [Timeframe.M15, Timeframe.H4]
        )
    config = PlatformConfig()
    config.primary_timeframe = timeframe
    config.confirmation_timeframes = confirmation_tfs
    aggregator = SignalAggregator(config)
    aggregator.data_provider = _get_shared_dp()
    return aggregator.analyze_pair(symbol=symbol)


async def _run_signal(
    symbol: str,
    timeframe: Timeframe,
    confirmation_tfs: Optional[List[Timeframe]] = None,
) -> SignalReport:
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(
        _executor, _run_signal_sync, symbol, timeframe, confirmation_tfs
    )


# ── Serialization ─────────────────────────────────────────────────────

def _serialize_report(report: SignalReport) -> dict:
    indicators = {}
    if report.primary_analysis:
        for name, ind in report.primary_analysis.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            indicators[name] = {
                "signal": ind.signal.name,
                "value": round(ind.value, 4),
                "confidence": round(ind.confidence, 1),
                "description": ind.description,
            }

    confirmation_data = []
    for ca in report.confirmation_analyses:
        cf_indicators = {}
        for name, ind in ca.indicators.items():
            if name.startswith("_") or ind.weight == 0:
                continue
            cf_indicators[name] = {
                "signal": ind.signal.name,
                "value": round(ind.value, 4),
                "confidence": round(ind.confidence, 1),
                "description": ind.description,
            }
        confirmation_data.append({
            "timeframe": ca.timeframe,
            "signal": ca.composite_signal.name,
            "score": round(ca.composite_score, 1),
            "confidence": round(ca.confidence, 1),
            "regime": ca.market_regime,
            "indicators": cf_indicators,
        })

    trade = None
    if report.trade_setup and report.final_signal not in (SignalStrength.NEUTRAL,):
        ts = report.trade_setup
        trade = {
            "direction": ts.direction,
            "entry": round(ts.entry_price, 5),
            "sl": round(ts.stop_loss, 5),
            "tp1": round(ts.take_profit_1, 5),
            "tp2": round(ts.take_profit_2, 5) if ts.take_profit_2 else None,
            "rr": round(ts.risk_reward, 2),
            "risk_pips": round(ts.risk_pips, 1),
            "reward_pips": round(ts.reward_pips, 1),
        }

    return {
        "pair": report.display_name,
        "symbol": report.symbol,
        "timestamp": report.timestamp,
        "signal": report.final_signal.name,
        "score": round(report.final_score, 1),
        "confidence": round(report.final_confidence, 1),
        "quality": report.signal_quality,
        "regime": report.market_regime,
        "volatility": report.volatility_regime,
        "price": report.current_price,
        "mtf_agreement": report.mtf_agreement,
        "mtf_signals": {k: v.name for k, v in report.mtf_signals.items()},
        "mtf_scores": {k: round(v, 1) for k, v in report.mtf_scores.items()},
        "indicators": indicators,
        "confirmations": confirmation_data,
        "trade_setup": trade,
        "bullish_reasons": report.bullish_reasons,
        "bearish_reasons": report.bearish_reasons,
        "warnings": report.warnings,
        "action": report.action_recommendation,
    }


# ── Redis publish ─────────────────────────────────────────────────────

def _publish_to_redis(snapshot: dict) -> None:
    payload = json.dumps(snapshot)
    _redis.set(REDIS_KEY, payload)
    _redis.publish(REDIS_CHANNEL, payload)


# ── Scan logic ────────────────────────────────────────────────────────

_signal_store: Dict[str, Dict[str, dict]] = {}


async def _scan_timeframe(tf: Timeframe, confirmation_tfs: List[Timeframe]) -> None:
    tasks = [_run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.exception(f"Scan failed for {SCAN_PAIRS[i]} {tf.value}: {result}")
            continue

        report: SignalReport = result
        display = report.display_name
        if display not in _signal_store:
            _signal_store[display] = {}
        _signal_store[display][tf.value] = _serialize_report(report)

    snapshot = {
        "type": "update",
        "timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC"),
        "data": _signal_store,
    }
    _publish_to_redis(snapshot)
    ok = len([r for r in results if not isinstance(r, Exception)])
    logger.info(f"{tf.value} scan complete — {ok} pairs updated")


# ── Candle-boundary scheduler ────────────────────────────────────────

def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
    now = datetime.now(timezone.utc)
    total_minutes = now.hour * 60 + now.minute
    elapsed_minutes = total_minutes % interval_minutes
    elapsed_seconds = elapsed_minutes * 60 + now.second + now.microsecond / 1e6
    interval_seconds = interval_minutes * 60
    remaining = interval_seconds - elapsed_seconds + buffer
    if remaining <= buffer:
        remaining += interval_seconds
    return remaining


async def _candle_loop(tf: Timeframe, interval_minutes: int) -> None:
    label = tf.value
    try:
        while True:
            wait = _seconds_until_boundary(interval_minutes, buffer=10)
            logger.info(f"{label} scan: next in {wait:.0f}s")
            await asyncio.sleep(wait)
            logger.info(f"{label} scan: firing")
            await _scan_timeframe(tf, AUTOSCAN_CONFIRMATIONS[tf])
    except asyncio.CancelledError:
        pass
    finally:
        logger.info(f"{label} scan loop stopped")


# ── Main ──────────────────────────────────────────────────────────────

async def main() -> None:
    _init_redis()

    logger.info("Running initial M15 scan…")
    await _scan_timeframe(Timeframe.M15, AUTOSCAN_CONFIRMATIONS[Timeframe.M15])
    logger.info("Initial scan complete")

    tasks = []
    for tf, interval in AUTOSCAN_LOOPS:
        task = asyncio.create_task(_candle_loop(tf, interval))
        tasks.append(task)
        logger.info(f"Started {tf.value} scan loop (every {interval}min)")

    stop = asyncio.Event()
    loop = asyncio.get_event_loop()
    for sig in (unix_signal.SIGINT, unix_signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)

    logger.info("Signal daemon running — Ctrl+C to stop")
    await stop.wait()

    logger.info("Shutting down…")
    for task in tasks:
        task.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    _executor.shutdown(wait=False)
    logger.info("Signal daemon stopped")


if __name__ == "__main__":
    asyncio.run(main())