from __future__ import annotations

import html
import os
import re
import sqlite3
import subprocess
import threading
from contextlib import closing
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Protocol

import requests
from dotenv import dotenv_values


TELEGRAM_TOKEN_PATTERN = re.compile(r"^[0-9]{6,12}:[A-Za-z0-9_-]{30,}$")
TELEGRAM_CHAT_PATTERN = re.compile(r"^-?[0-9]{4,20}$")


def as_decimal(value: object) -> Decimal:
    try:
        return Decimal(str(value))
    except (InvalidOperation, TypeError, ValueError):
        return Decimal("0")


def utc(value: datetime | None = None) -> datetime:
    return (value or datetime.now(timezone.utc)).astimezone(timezone.utc)


class NotificationChannel(Protocol):
    name: str

    def send(self, title: str, body: str) -> bool: ...


class NotificationStore:
    def __init__(self, database_file: Path) -> None:
        self.database_file = Path(database_file)
        self._lock = threading.Lock()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(self.database_file, timeout=15)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA busy_timeout = 15000")
        connection.execute("PRAGMA journal_mode = WAL")
        connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS notification_deliveries (
                event_id TEXT NOT NULL,
                channel TEXT NOT NULL,
                status TEXT NOT NULL CHECK (status IN ('SENT', 'FAILED')),
                attempt_count INTEGER NOT NULL,
                last_attempt_at TEXT NOT NULL,
                delivered_at TEXT,
                error_code TEXT,
                PRIMARY KEY(event_id, channel)
            );
            CREATE TABLE IF NOT EXISTS notification_meta (
                key TEXT PRIMARY KEY,
                value TEXT NOT NULL
            );
            """
        )
        return connection

    def can_attempt(
        self,
        event_id: str,
        channel: str,
        *,
        now: datetime,
        retry_minutes: int = 10,
        max_attempts: int = 3,
    ) -> bool:
        with self._lock, closing(self._connect()) as connection:
            row = connection.execute(
                "SELECT * FROM notification_deliveries WHERE event_id = ? AND channel = ?",
                (event_id, channel),
            ).fetchone()
        if row is None:
            return True
        if row["status"] == "SENT" or int(row["attempt_count"]) >= max_attempts:
            return False
        last_attempt = datetime.fromisoformat(str(row["last_attempt_at"]))
        return (now - last_attempt).total_seconds() >= retry_minutes * 60

    def record(self, event_id: str, channel: str, sent: bool, *, now: datetime) -> None:
        now_text = now.isoformat()
        with self._lock, closing(self._connect()) as connection:
            connection.execute(
                """
                INSERT INTO notification_deliveries(
                    event_id, channel, status, attempt_count, last_attempt_at,
                    delivered_at, error_code
                ) VALUES (?, ?, ?, 1, ?, ?, ?)
                ON CONFLICT(event_id, channel) DO UPDATE SET
                    status = excluded.status,
                    attempt_count = notification_deliveries.attempt_count + 1,
                    last_attempt_at = excluded.last_attempt_at,
                    delivered_at = excluded.delivered_at,
                    error_code = excluded.error_code
                """,
                (
                    event_id,
                    channel,
                    "SENT" if sent else "FAILED",
                    now_text,
                    now_text if sent else None,
                    None if sent else "DELIVERY_FAILED",
                ),
            )
            connection.commit()

    def recent_deliveries(
        self, event_prefix: str, limit: int = 50
    ) -> list[dict[str, Any]]:
        """Most recent delivery rows whose event id starts with a prefix."""
        escaped = (
            event_prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        )
        with self._lock, closing(self._connect()) as connection:
            rows = connection.execute(
                """
                SELECT event_id, channel, status, delivered_at, last_attempt_at
                FROM notification_deliveries
                WHERE event_id LIKE ? ESCAPE '\\'
                ORDER BY COALESCE(delivered_at, last_attempt_at) DESC
                LIMIT ?
                """,
                (escaped + "%", max(1, int(limit))),
            ).fetchall()
        return [
            {
                "event_id": str(row["event_id"]),
                "channel": str(row["channel"]),
                "status": str(row["status"]),
                "delivered_at": row["delivered_at"],
            }
            for row in rows
        ]

    def note_scan_failure(self, *, now: datetime, stale_seconds: int = 180) -> str | None:
        now_text = now.isoformat()
        with self._lock, closing(self._connect()) as connection:
            rows = connection.execute(
                "SELECT key, value FROM notification_meta WHERE key IN ('stale_since', 'stale_alerted')"
            ).fetchall()
            meta = {str(row["key"]): str(row["value"]) for row in rows}
            stale_since = meta.get("stale_since")
            if stale_since is None:
                connection.execute(
                    "INSERT OR REPLACE INTO notification_meta(key, value) VALUES ('stale_since', ?)",
                    (now_text,),
                )
                connection.execute(
                    "INSERT OR REPLACE INTO notification_meta(key, value) VALUES ('stale_alerted', '0')"
                )
                connection.commit()
                return None
            if meta.get("stale_alerted") == "1":
                return None
            started = datetime.fromisoformat(stale_since)
            if (now - started).total_seconds() < stale_seconds:
                return None
            connection.execute(
                "INSERT OR REPLACE INTO notification_meta(key, value) VALUES ('stale_alerted', '1')"
            )
            connection.commit()
            return stale_since

    def note_scan_fresh(self) -> None:
        with self._lock, closing(self._connect()) as connection:
            connection.execute(
                "DELETE FROM notification_meta WHERE key IN ('stale_since', 'stale_alerted')"
            )
            connection.commit()

    def get_meta(self, key: str) -> str | None:
        with self._lock, closing(self._connect()) as connection:
            row = connection.execute(
                "SELECT value FROM notification_meta WHERE key = ?", (key,)
            ).fetchone()
        return str(row["value"]) if row is not None else None

    def set_meta(self, key: str, value: str | None) -> None:
        with self._lock, closing(self._connect()) as connection:
            if value is None:
                connection.execute("DELETE FROM notification_meta WHERE key = ?", (key,))
            else:
                connection.execute(
                    "INSERT OR REPLACE INTO notification_meta(key, value) VALUES (?, ?)",
                    (key, value),
                )
            connection.commit()


class WindowsToastNotifier:
    name = "WINDOWS"

    def send(self, title: str, body: str) -> bool:
        if os.name != "nt":
            return False
        safe_title = html.escape(title, quote=True)
        safe_body = html.escape(body, quote=True)
        script = f"""
[Windows.UI.Notifications.ToastNotificationManager, Windows.UI.Notifications, ContentType = WindowsRuntime] | Out-Null
[Windows.Data.Xml.Dom.XmlDocument, Windows.Data.Xml.Dom.XmlDocument, ContentType = WindowsRuntime] | Out-Null
$xml = New-Object Windows.Data.Xml.Dom.XmlDocument
$xml.LoadXml('<toast><visual><binding template="ToastGeneric"><text>{safe_title}</text><text>{safe_body}</text></binding></visual></toast>')
$toast = [Windows.UI.Notifications.ToastNotification]::new($xml)
[Windows.UI.Notifications.ToastNotificationManager]::CreateToastNotifier('Binance Spot Assistant').Show($toast)
"""
        encoded = script.encode("utf-16-le")
        import base64

        command = [
            "powershell.exe",
            "-NoProfile",
            "-NonInteractive",
            "-EncodedCommand",
            base64.b64encode(encoded).decode("ascii"),
        ]
        try:
            completed = subprocess.run(
                command,
                capture_output=True,
                timeout=10,
                check=False,
                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
            )
        except (OSError, subprocess.SubprocessError):
            return False
        return completed.returncode == 0


class TelegramNotifier:
    name = "TELEGRAM"

    def __init__(self, token: str, chat_id: str, session: requests.Session | None = None) -> None:
        if not TELEGRAM_TOKEN_PATTERN.fullmatch(token) or not TELEGRAM_CHAT_PATTERN.fullmatch(chat_id):
            raise ValueError("Telegram notification settings are invalid.")
        self._token = token
        self._chat_id = chat_id
        self._session = session or requests.Session()

    def send(self, title: str, body: str) -> bool:
        try:
            response = self._session.post(
                f"https://api.telegram.org/bot{self._token}/sendMessage",
                json={
                    "chat_id": self._chat_id,
                    "text": f"{title}\n\n{body}",
                    "disable_web_page_preview": True,
                },
                timeout=10,
            )
        except requests.RequestException:
            return False
        return response.status_code == 200


def strong_notification_candidate(
    analysis: dict[str, Any],
    *,
    min_score: Decimal = Decimal("0.70"),
    min_confidence: Decimal = Decimal("80"),
    max_stop_risk: Decimal = Decimal("7"),
    min_reward_risk: Decimal = Decimal("2"),
    require_new: bool = True,
) -> bool:
    levels = analysis.get("levels") if isinstance(analysis.get("levels"), dict) else {}
    risk = analysis.get("risk") if isinstance(analysis.get("risk"), dict) else {}
    gates = analysis.get("quality_gates") if isinstance(analysis.get("quality_gates"), dict) else {}
    structure = (
        analysis.get("support_resistance")
        if isinstance(analysis.get("support_resistance"), dict)
        else {}
    )
    current = as_decimal(analysis.get("current_price"))
    entry_low = as_decimal(levels.get("entry_low"))
    entry_high = as_decimal(levels.get("entry_high"))
    return bool(
        analysis.get("decision") == "BUY_SETUP"
        and (analysis.get("notification_eligible") is True or not require_new)
        and as_decimal(analysis.get("combined_score")) >= min_score
        and as_decimal(analysis.get("confidence_percent")) >= min_confidence
        and entry_low > 0
        and entry_low <= current <= entry_high
        and Decimal("0") < as_decimal(risk.get("stop_risk_percent")) <= max_stop_risk
        and as_decimal(risk.get("reward_risk_target_1")) >= min_reward_risk
        and gates
        and all(value is True for value in gates.values())
        and structure.get("fake_breakout") is not True
        and structure.get("late_entry") is not True
    )


def opportunity_rank(analysis: dict[str, Any]) -> tuple[Decimal, ...]:
    risk = analysis.get("risk") if isinstance(analysis.get("risk"), dict) else {}
    timeframe_items = analysis.get("timeframes") if isinstance(analysis.get("timeframes"), list) else []
    volume = max(
        (as_decimal(item.get("volume_ratio")) for item in timeframe_items if isinstance(item, dict)),
        default=Decimal("0"),
    )
    return (
        as_decimal(analysis.get("combined_score")),
        as_decimal(analysis.get("confidence_percent")),
        as_decimal(risk.get("reward_risk_target_1")),
        -as_decimal(risk.get("stop_risk_percent")),
        volume,
    )


def select_best_opportunity(
    analyses: list[dict[str, Any]],
    *,
    min_score: Decimal = Decimal("0.70"),
    min_confidence: Decimal = Decimal("80"),
    max_stop_risk: Decimal = Decimal("7"),
    min_reward_risk: Decimal = Decimal("2"),
) -> dict[str, Any] | None:
    candidates = [
        analysis
        for analysis in analyses
        if isinstance(analysis.get("signal"), dict)
        and strong_notification_candidate(
            analysis,
            min_score=min_score,
            min_confidence=min_confidence,
            max_stop_risk=max_stop_risk,
            min_reward_risk=min_reward_risk,
            require_new=False,
        )
    ]
    return max(candidates, key=opportunity_rank, default=None)


READINESS_WEIGHTS: dict[str, int] = {
    "required_timeframes_present": 3,
    "higher_timeframes_bullish": 16,
    "one_hour_entry_ready": 6,
    "combined_score": 10,
    "confidence": 8,
    "volume_confirmed": 8,
    "price_inside_entry": 12,
    "liquidity_confirmed": 5,
    "daily_move_acceptable": 3,
    "market_favorable": 8,
    "no_fake_breakout": 4,
    "not_late_entry": 5,
    "volatility_acceptable": 3,
    "stop_risk": 4,
    "reward_risk": 3,
    "resistance_clear": 2,
}


def opportunity_readiness(
    analysis: dict[str, Any],
    *,
    min_score: Decimal = Decimal("0.70"),
    min_confidence: Decimal = Decimal("80"),
    max_stop_risk: Decimal = Decimal("7"),
    min_reward_risk: Decimal = Decimal("2"),
) -> dict[str, Any] | None:
    """Return a safe watch candidate and its rule-completion score.

    The percentage measures completion of this project's rules. It is not a
    probability of profit. Unsafe or materially late candidates are omitted.
    """
    if analysis.get("decision") in {"SELL_REVIEW", "ERROR"}:
        return None

    timeframes = {
        str(item.get("interval")): item
        for item in analysis.get("timeframes", [])
        if isinstance(item, dict)
    }
    required_present = all(interval in timeframes for interval in ("1h", "4h", "1d"))
    if not required_present:
        return None
    if any(timeframes[interval].get("trend") != "BULLISH" for interval in ("4h", "1d")):
        return None
    one_hour = timeframes["1h"]
    one_hour_ready = one_hour.get("trend") == "BULLISH" or (
        one_hour.get("trend") == "MIXED" and as_decimal(one_hour.get("score")) >= 0
    )
    if one_hour.get("trend") == "BEARISH":
        return None

    levels = analysis.get("levels") if isinstance(analysis.get("levels"), dict) else {}
    risk = analysis.get("risk") if isinstance(analysis.get("risk"), dict) else {}
    gates = (
        analysis.get("quality_gates")
        if isinstance(analysis.get("quality_gates"), dict)
        else {}
    )
    structure = (
        analysis.get("support_resistance")
        if isinstance(analysis.get("support_resistance"), dict)
        else {}
    )
    current = as_decimal(analysis.get("current_price"))
    entry_low = as_decimal(levels.get("entry_low"))
    entry_high = as_decimal(levels.get("entry_high"))
    stop_risk = as_decimal(risk.get("stop_risk_percent"))
    reward_risk = as_decimal(risk.get("reward_risk_target_1"))
    if current <= 0 or entry_low <= 0 or entry_high < entry_low:
        return None
    if not Decimal("0") < stop_risk <= max_stop_risk:
        return None
    if gates.get("liquidity_confirmed") is not True:
        return None
    if gates.get("daily_move_acceptable") is not True:
        return None
    if gates.get("volatility_acceptable") is not True:
        return None
    if structure.get("fake_breakout") is True or gates.get("no_fake_breakout") is False:
        return None

    if current > entry_high:
        distance_percent = ((current - entry_high) / entry_high) * Decimal("100")
    elif current < entry_low:
        distance_percent = ((entry_low - current) / entry_low) * Decimal("100")
    else:
        distance_percent = Decimal("0")
    if distance_percent > Decimal("3"):
        return None

    conditions = {
        "required_timeframes_present": required_present,
        "higher_timeframes_bullish": True,
        "one_hour_entry_ready": one_hour_ready,
        "combined_score": as_decimal(analysis.get("combined_score")) >= min_score,
        "confidence": as_decimal(analysis.get("confidence_percent")) >= min_confidence,
        "volume_confirmed": gates.get("volume_confirmed") is True,
        "price_inside_entry": entry_low <= current <= entry_high,
        "liquidity_confirmed": True,
        "daily_move_acceptable": True,
        "market_favorable": gates.get("market_favorable") is True,
        "no_fake_breakout": True,
        "not_late_entry": gates.get("not_late_entry") is True,
        "volatility_acceptable": True,
        "stop_risk": True,
        "reward_risk": reward_risk >= min_reward_risk,
        "resistance_clear": gates.get("resistance_clear") is True,
    }
    readiness = sum(
        weight for name, weight in READINESS_WEIGHTS.items() if conditions.get(name) is True
    )
    missing = [name for name in READINESS_WEIGHTS if conditions.get(name) is not True]
    return {
        "symbol": analysis.get("symbol"),
        "status": "READY" if readiness == 100 else ("NEAR" if readiness >= 85 else "WATCH"),
        "readiness_percent": readiness,
        "current_price": analysis.get("current_price"),
        "combined_score": analysis.get("combined_score"),
        "confidence_percent": analysis.get("confidence_percent"),
        "levels": levels,
        "risk": risk,
        "missing_conditions": missing,
        "distance_to_entry_percent": format(distance_percent, ".2f"),
    }


def select_watch_opportunities(
    analyses: list[dict[str, Any]],
    *,
    exclude_symbol: str | None = None,
    limit: int = 3,
    min_readiness: int = 65,
    min_score: Decimal = Decimal("0.70"),
    min_confidence: Decimal = Decimal("80"),
    max_stop_risk: Decimal = Decimal("7"),
    min_reward_risk: Decimal = Decimal("2"),
) -> list[dict[str, Any]]:
    options: list[dict[str, Any]] = []
    for analysis in analyses:
        if exclude_symbol and analysis.get("symbol") == exclude_symbol:
            continue
        option = opportunity_readiness(
            analysis,
            min_score=min_score,
            min_confidence=min_confidence,
            max_stop_risk=max_stop_risk,
            min_reward_risk=min_reward_risk,
        )
        if option is not None and option["readiness_percent"] >= min_readiness:
            options.append(option)
    options.sort(
        key=lambda item: (
            int(item["readiness_percent"]),
            as_decimal(item.get("combined_score")),
            as_decimal(item.get("confidence_percent")),
        ),
        reverse=True,
    )
    return options[: max(0, limit)]


# Gates that only describe price timing inside an already-qualified setup. When
# every other gate passes, the setup itself is sound and the only question is
# whether the price is offering an entry right now.
ENTRY_ZONE_TIMING_GATES = {"price_inside_entry", "not_late_entry"}


def entry_zone_state(
    analysis: dict[str, Any],
    *,
    near_percent: Decimal = Decimal("0.5"),
) -> str | None:
    """Classify price position relative to the entry zone.

    Returns "INSIDE" when the live price is within the entry range, "NEAR"
    when it is within ``near_percent`` of the zone edge, or None when the
    setup is not sound enough to chase or the price is far away. A BUY_SETUP
    decision is excluded because the strong-signal notification covers it.
    """
    if analysis.get("decision") == "BUY_SETUP":
        return None
    levels = analysis.get("levels") if isinstance(analysis.get("levels"), dict) else {}
    gates = (
        analysis.get("quality_gates")
        if isinstance(analysis.get("quality_gates"), dict)
        else {}
    )
    entry_low = as_decimal(levels.get("entry_low"))
    entry_high = as_decimal(levels.get("entry_high"))
    price = as_decimal(analysis.get("current_price"))
    if entry_low <= 0 or entry_high < entry_low or price <= 0 or not gates:
        return None
    failed = {str(name) for name, passed in gates.items() if passed is not True}
    if any(name not in ENTRY_ZONE_TIMING_GATES for name in failed):
        return None
    if not failed:
        # All gates pass without BUY_SETUP: let the normal lifecycle handle it.
        return None
    if entry_low <= price <= entry_high:
        return "INSIDE"
    edge = entry_high if price > entry_high else entry_low
    if edge > 0 and ((abs(price - edge) / edge) * Decimal("100")) <= near_percent:
        return "NEAR"
    return None


def arabic_entry_zone_message(
    analysis: dict[str, Any],
    state: str,
) -> tuple[str, str]:
    levels = analysis["levels"]
    symbol = str(analysis.get("symbol", ""))
    price = analysis.get("current_price")
    if state == "INSIDE":
        title = f"السعر داخل منطقة الدخول الآن: {symbol}"
        body = (
            f"سعر {symbol} الحالي {price} داخل نطاق الدخول "
            f"{levels.get('entry_low')} - {levels.get('entry_high')}.\n"
            f"وقف الخسارة: {levels.get('stop_loss')}\n"
            f"الهدف الأول: {levels.get('target_1')} / الثاني: {levels.get('target_2')}\n\n"
            "هذه فرصة دخول محتملة — قرر بنفسك وبمبلغ صغير والتزم بالوقف. لم يُرسل أي أمر."
        )
    else:
        title = f"اقتراب من منطقة دخول {symbol}"
        body = (
            f"سعر {symbol} الحالي {price} قريب جدًا من نطاق الدخول "
            f"{levels.get('entry_low')} - {levels.get('entry_high')}.\n"
            "راقب الشاشة؛ إذا دخل السعر النطاق واكتملت الشروط ستصلك إشارة قوية. لم يُرسل أي أمر."
        )
    return title, body


def arabic_opportunity_message(analysis: dict[str, Any]) -> tuple[str, str]:
    levels = analysis["levels"]
    risk = analysis["risk"]
    symbol = str(analysis.get("symbol", ""))
    title = f"فرصة قوية للتجربة اليدوية: {symbol}"
    body = (
        f"السعر: {analysis.get('current_price')}\n"
        f"نطاق الدخول: {levels.get('entry_low')} - {levels.get('entry_high')}\n"
        f"وقف الخسارة: {levels.get('stop_loss')}\n"
        f"الهدف الأول: {levels.get('target_1')}\n"
        f"الهدف الثاني: {levels.get('target_2')}\n"
        f"الثقة: {analysis.get('confidence_percent')}%\n"
        f"مخاطرة الوقف: {risk.get('stop_risk_percent')}%\n"
        f"العائد/المخاطرة: {risk.get('reward_risk_target_1')}:1\n\n"
        "أنصح بدراسة هذه الفرصة للتجربة اليدوية بحجم صغير لأنها اجتازت جميع شروط الجودة الصارمة. "
        "لم يُرسل أي أمر شراء أو بيع."
    )
    return title, body


class NotificationManager:
    def __init__(self, store: NotificationStore, secret_file: Path) -> None:
        self.store = store
        self.secret_file = Path(secret_file)

    def _telegram_credentials(self) -> tuple[str, str] | None:
        if not self.secret_file.is_file():
            return None
        values = dotenv_values(self.secret_file)
        token = str(values.get("TELEGRAM_BOT_TOKEN") or "").strip()
        chat_id = str(values.get("TELEGRAM_CHAT_ID") or "").strip()
        if not TELEGRAM_TOKEN_PATTERN.fullmatch(token) or not TELEGRAM_CHAT_PATTERN.fullmatch(chat_id):
            return None
        return token, chat_id

    def public_status(self, config: dict[str, Any]) -> dict[str, bool]:
        telegram_requested = config.get("telegram_notifications_enabled") is True
        return {
            "windows_enabled": config.get("windows_notifications_enabled") is True,
            "telegram_requested": telegram_requested,
            "telegram_configured": telegram_requested and self._telegram_credentials() is not None,
        }

    def _channels(self, config: dict[str, Any]) -> list[NotificationChannel]:
        channels: list[NotificationChannel] = []
        if config.get("windows_notifications_enabled") is True:
            channels.append(WindowsToastNotifier())
        credentials = self._telegram_credentials()
        if config.get("telegram_notifications_enabled") is True and credentials is not None:
            channels.append(TelegramNotifier(*credentials))
        return channels

    def _deliver(
        self,
        event_id: str,
        title: str,
        body: str,
        config: dict[str, Any],
        *,
        now: datetime,
    ) -> dict[str, int]:
        summary = {"sent": 0, "failed": 0, "skipped": 0}
        for channel in self._channels(config):
            if not self.store.can_attempt(event_id, channel.name, now=now):
                summary["skipped"] += 1
                continue
            sent = channel.send(title, body)
            self.store.record(event_id, channel.name, sent, now=now)
            summary["sent" if sent else "failed"] += 1
        return summary

    def process(
        self,
        analyses: list[dict[str, Any]],
        signal_state: dict[str, Any],
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        summary = {"sent": 0, "failed": 0, "skipped": 0, "qualified": 0}
        if config.get("notifications_enabled") is not True:
            return summary
        current = utc(now)
        active_by_symbol = {
            str(item.get("symbol", "")): str(item.get("id", ""))
            for item in signal_state.get("active_signals", [])
        }
        strict_candidates = []
        for analysis in analyses:
            symbol = str(analysis.get("symbol", ""))
            signal_id = active_by_symbol.get(symbol)
            if not signal_id:
                continue
            candidate = dict(analysis)
            candidate["signal"] = {"id": signal_id, "status": "ACTIVE"}
            if strong_notification_candidate(
                candidate,
                min_score=as_decimal(config.get("notification_min_score", "0.70")),
                min_confidence=as_decimal(config.get("notification_min_confidence_percent", "80")),
                max_stop_risk=as_decimal(config.get("notification_max_stop_risk_percent", "7")),
                min_reward_risk=as_decimal(config.get("notification_min_reward_risk", "2")),
                require_new=False,
            ):
                strict_candidates.append(candidate)

        summary["qualified"] = len(strict_candidates)
        current_id = self.store.get_meta("current_recommendation_signal_id")
        strict_ids = {str(item["signal"]["id"]) for item in strict_candidates}
        selected = next(
            (
                item
                for item in strict_candidates
                if str(item["signal"]["id"]) == current_id
            ),
            None,
        )
        if current_id not in strict_ids:
            self.store.set_meta("current_recommendation_signal_id", None)
            selected = max(strict_candidates, key=opportunity_rank, default=None)
        if selected is not None:
            event_id = str(selected["signal"]["id"])
            self.store.set_meta("current_recommendation_signal_id", event_id)
            title, body = arabic_opportunity_message(selected)
            result = self._deliver(event_id, title, body, config, now=current)
            for key in ("sent", "failed", "skipped"):
                summary[key] += result[key]
        return summary

    def process_price_alerts(
        self,
        market: list[dict[str, Any]],
        monitors: list[dict[str, Any]],
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        """Deliver each configured price threshold once; never place an order."""
        summary = {"sent": 0, "failed": 0, "skipped": 0, "triggered": 0}
        if config.get("notifications_enabled") is not True:
            return summary
        current = utc(now)
        prices = {
            str(item.get("symbol", "")).strip().upper(): as_decimal(item.get("price"))
            for item in market
        }
        for monitor in monitors:
            symbol = str(monitor.get("symbol", "")).strip().upper()
            monitor_id = str(monitor.get("id", "")).strip()
            price = prices.get(symbol, Decimal("0"))
            if not symbol or not monitor_id or price <= 0:
                continue
            thresholds = (
                ("ABOVE", as_decimal(monitor.get("alert_above")), price >= as_decimal(monitor.get("alert_above"))),
                ("BELOW", as_decimal(monitor.get("alert_below")), price <= as_decimal(monitor.get("alert_below"))),
            )
            for direction, threshold, reached in thresholds:
                if threshold <= 0 or not reached:
                    continue
                summary["triggered"] += 1
                direction_ar = "الصعود" if direction == "ABOVE" else "الهبوط"
                title = f"تنبيه سعر {symbol}"
                body = (
                    f"وصل سعر {symbol} إلى {price} بعد تنبيه {direction_ar} عند {threshold}.\n"
                    "هذا تنبيه سعري فقط وليس توصية شراء أو بيع. لم يُرسل أي أمر."
                )
                result = self._deliver(
                    f"PRICE:{monitor_id}:{direction}",
                    title,
                    body,
                    config,
                    now=current,
                )
                for key in ("sent", "failed", "skipped"):
                    summary[key] += result[key]
        return summary

    def process_favorite_alerts(
        self,
        market: list[dict[str, Any]],
        favorite_alerts: dict[str, Any],
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        """Deliver favorite-coin price thresholds once per (symbol, side, level).

        The threshold is embedded in the event id, so editing a target re-arms
        that side automatically while an unchanged target never re-fires.
        """
        summary = {"sent": 0, "failed": 0, "skipped": 0, "triggered": 0}
        if config.get("notifications_enabled") is not True:
            return summary
        current = utc(now)
        prices = {
            str(item.get("symbol", "")).strip().upper(): as_decimal(item.get("price"))
            for item in market
        }
        for symbol_raw, alert in (favorite_alerts or {}).items():
            symbol = str(symbol_raw).strip().upper()
            if not symbol or not isinstance(alert, dict):
                continue
            price = prices.get(symbol, Decimal("0"))
            if price <= 0:
                continue
            thresholds = (
                ("ABOVE", as_decimal(alert.get("above")), price >= as_decimal(alert.get("above"))),
                ("BELOW", as_decimal(alert.get("below")), price <= as_decimal(alert.get("below"))),
            )
            for direction, threshold, reached in thresholds:
                if threshold <= 0 or not reached:
                    continue
                summary["triggered"] += 1
                direction_ar = "الصعود" if direction == "ABOVE" else "الهبوط"
                title = f"تنبيه سعر {symbol}"
                body = (
                    f"وصل سعر {symbol} إلى {price} بعد تنبيه {direction_ar} عند {threshold}.\n"
                    "هذا تنبيه سعري فقط وليس توصية شراء أو بيع. لم يُرسل أي أمر."
                )
                result = self._deliver(
                    f"PRICEFAV:{symbol}:{direction}:{threshold}",
                    title,
                    body,
                    config,
                    now=current,
                )
                for key in ("sent", "failed", "skipped"):
                    summary[key] += result[key]
        return summary

    def send_once(
        self,
        event_id: str,
        title: str,
        body: str,
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        """Deliver a one-off guardrail message with per-channel dedup."""
        return self._deliver(event_id, title, body, config, now=utc(now))

    def process_entry_zone_alerts(
        self,
        analyses: list[dict[str, Any]],
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        """Notify once per day when price enters or nears a sound entry zone.

        Covers the gap where a setup is qualified on every non-price gate but
        the price is not inside the entry range yet — the historical case that
        invalidated most strong signals as late entries.
        """
        summary = {"sent": 0, "failed": 0, "skipped": 0, "inside": 0, "near": 0}
        if config.get("entry_zone_alerts_enabled", True) is not True:
            return summary
        if config.get("notifications_enabled") is not True:
            return summary
        current = utc(now)
        near_percent = as_decimal(config.get("entry_zone_near_percent", "0.5"))
        for analysis in analyses:
            state = entry_zone_state(analysis, near_percent=near_percent)
            if state is None:
                continue
            symbol = str(analysis.get("symbol", "")).strip().upper()
            if not symbol:
                continue
            summary["inside" if state == "INSIDE" else "near"] += 1
            event_id = f"ENTRYZONE:{symbol}:{state}:{current.date().isoformat()}"
            if not any(
                self.store.can_attempt(event_id, channel.name, now=current)
                for channel in self._channels(config)
            ):
                summary["skipped"] += 1
                continue
            title, body = arabic_entry_zone_message(analysis, state)
            result = self._deliver(event_id, title, body, config, now=current)
            for key in ("sent", "failed", "skipped"):
                summary[key] += result[key]
        return summary

    def process_daily_portfolio_summary(
        self,
        trade_blocks: list[dict[str, Any]],
        config: dict[str, Any],
        *,
        now: datetime | None = None,
    ) -> dict[str, int]:
        """Send one morning notification with open-position P&L totals per local day.

        Fires on the first dashboard refresh after ``daily_portfolio_summary_hour``
        (local time, default 08:00). The event id embeds the local date, and the
        delivery store dedups per channel, so it is delivered exactly once a day.
        """
        summary = {"sent": 0, "failed": 0, "skipped": 0, "positions": 0}
        if config.get("daily_portfolio_summary_enabled", True) is not True:
            return summary
        if config.get("notifications_enabled") is not True:
            return summary
        current = utc(now).astimezone()
        try:
            hour_gate = int(config.get("daily_portfolio_summary_hour", 8))
        except (TypeError, ValueError):
            hour_gate = 8
        if current.hour < max(0, min(hour_gate, 23)):
            return summary
        positions: list[tuple[str, Decimal, Decimal, Decimal, Decimal, Decimal, Decimal | None]] = []
        for block in trade_blocks or []:
            open_pos = block.get("open_position")
            if not isinstance(open_pos, dict):
                continue
            quantity = as_decimal(open_pos.get("quantity"))
            cost = as_decimal(open_pos.get("cost_usdt"))
            average_buy = as_decimal(open_pos.get("average_buy_price"))
            price = as_decimal(block.get("current_price"))
            if quantity <= 0 or cost <= 0:
                continue
            value = quantity * price if price > 0 else Decimal("0")
            pnl = (value - cost) if price > 0 else None
            positions.append(
                (str(block.get("symbol", "")).strip().upper(), quantity, average_buy, price, cost, value, pnl)
            )
        summary["positions"] = len(positions)
        event_id = f"DAILY:PORTFOLIO:{current.date().isoformat()}"
        title = f"ملخص محفظتك اليومي — {current.date().isoformat()}"
        if not positions:
            body = (
                "لا توجد مراكز مفتوحة في محفظتك حاليًا.\n\n"
                "ملخص يومي للمتابعة فقط — لم يُرسل أي أمر شراء أو بيع."
            )
        else:
            total_cost = sum(item[4] for item in positions)
            total_value = sum(item[5] for item in positions)
            total_pnl = total_value - total_cost
            total_pct = (total_pnl / total_cost * 100) if total_cost > 0 else Decimal("0")
            lines = []
            for symbol, quantity, average_buy, price, cost, _value, pnl in positions:
                if pnl is None or cost <= 0:
                    pnl_text = "—"
                else:
                    pct = pnl / cost * 100
                    pnl_text = f"{pnl:+.2f} USDT ({pct:+.2f}%)"
                price_text = str(price) if price > 0 else "—"
                lines.append(
                    f"{symbol}: {quantity} @ متوسط {average_buy} ← الآن {price_text} · {pnl_text}"
                )
            body = (
                "المراكز المفتوحة:\n"
                + "\n".join(lines)
                + "\n\n"
                + f"إجمالي التكلفة: {total_cost:.2f} USDT\n"
                + f"القيمة الحالية: {total_value:.2f} USDT\n"
                + f"الربح/الخسارة: {total_pnl:+.2f} USDT ({total_pct:+.2f}%)\n\n"
                + "ملخص يومي للمتابعة فقط — لم يُرسل أي أمر شراء أو بيع."
            )
        result = self._deliver(event_id, title, body, config, now=current)
        for key in ("sent", "failed", "skipped"):
            summary[key] += result[key]
        return summary

    def scan_failed(self, config: dict[str, Any], *, now: datetime | None = None) -> None:
        current = utc(now)
        stale_since = self.store.note_scan_failure(now=current, stale_seconds=180)
        if stale_since is None or config.get("notifications_enabled") is not True:
            return
        self._deliver(
            f"STALE:{stale_since}",
            "توقف تحديث بيانات Binance Spot",
            "أصبحت البيانات أقدم من 3 دقائق. قد تكون نافذة التشغيل أو الشبكة أو اتصال Binance متوقفًا. لم يُرسل أي أمر.",
            config,
            now=current,
        )

    def scan_fresh(self) -> None:
        self.store.note_scan_fresh()