ExternalSignalStrategy

# user_data/strategies/ExternalSignalStrategy.py
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple

import json
import os
import requests
import pandas as pd  # noqa
from freqtrade.strategy import IStrategy, IntParameter, CategoricalParameter
from freqtrade.exchange import timeframe_to_minutes

class ExternalSignalStrategy(IStrategy):
    """
    Very simple strategy:
      - Reads an external API: {"buy": [...], "sell": [...]}
      - Sells first (signals on pairs in 'sell').
      - Buys on pairs in 'buy'.
      - Uses equal-share stake across current buy targets.
    """

    # --- Required minimal settings ---
    timeframe = "1m"
    startup_candle_count = 1

    # Basic risk placeholders (you can tweak or ignore — exits are signal-based)
    minimal_roi = {"0": 10_000}  # effectively disable ROI-based exits
    stoploss = -0.99             # rely on signal-based sell; you may raise this later
    trailing_stop = False

    # Cache for API results
    _signal_cache: Dict[str, any] = {}
    _signal_cache_expiry: Optional[datetime] = None

    # How often to refresh external API (seconds)
    refresh_seconds = 20

    # Config key or env var for the API URL
    config_key = "external_signal_api"
    env_key = "EXTERNAL_SIGNAL_API"

    # Optionally force quote (e.g. "USDT") if API sends tickers like "BTC"
    force_quote: Optional[str] = None  # e.g., "USDT"

    def _get_api_url(self) -> str:
        # Priority: config.json -> env var -> default
        url = self.config.get(self.config_key) or os.getenv(self.env_key)
        if not url:
            # Fallback useful for local tests
            url = "http://localhost:5000/signals"
        return url

    def _normalize_pair(self, pair: str) -> str:
        """
        Accepts: 'BTC/USDT', 'BTC-USDT', 'BTCUSDT', 'BTC'
        Optionally appends force_quote (e.g., 'USDT') if no quote detected.
        """
        p = pair.strip().upper().replace("-", "/")
        if "/" not in p:
            # Try to infer quote
            if self.force_quote:
                p = f"{p}/{self.force_quote}"
        return p

    def _fetch_signals(self) -> Tuple[List[str], List[str]]:
        """
        Fetch and cache external signals (buy/sell lists).
        Cache for self.refresh_seconds to avoid hammering the API.
        """
        now = datetime.utcnow()
        if self._signal_cache_expiry and now < self._signal_cache_expiry:
            buy = self._signal_cache.get("buy", [])
            sell = self._signal_cache.get("sell", [])
            return buy, sell

        url = self._get_api_url()
        try:
            r = requests.get(url, timeout=4)
            r.raise_for_status()
            data = r.json() if r.headers.get("Content-Type", "").startswith("application/json") else json.loads(r.text)
            raw_buy = data.get("buy", []) or []
            raw_sell = data.get("sell", []) or []

            buy = [self._normalize_pair(x) for x in raw_buy]
            sell = [self._normalize_pair(x) for x in raw_sell]

            self._signal_cache = {"buy": buy, "sell": sell}
            self._signal_cache_expiry = now + timedelta(seconds=self.refresh_seconds)

            return buy, sell
        except Exception as e:
            # On error, do nothing (empty lists) to be safe
            self.logger.warning(f"ExternalSignalStrategy: API fetch failed: {e}")
            self._signal_cache = {"buy": [], "sell": []}
            self._signal_cache_expiry = now + timedelta(seconds=self.refresh_seconds)
            return [], []

    # --- Signal population ---
    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # No indicators needed — signals are external
        return dataframe

    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        buy_list, sell_list = self._fetch_signals()
        pair = metadata["pair"].upper()

        dataframe["buy"] = 0
        # Only buy if this pair is in buy_list and NOT in sell_list (sell has priority)
        if pair in buy_list and pair not in sell_list and len(dataframe) > 0:
            dataframe.loc[dataframe.index[-1], "buy"] = 1
        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        buy_list, sell_list = self._fetch_signals()
        pair = metadata["pair"].upper()

        dataframe["sell"] = 0
        # Emit sell if pair is listed to sell
        if pair in sell_list and len(dataframe) > 0:
            dataframe.loc[dataframe.index[-1], "sell"] = 1
        return dataframe

    # --- Optional: block buys while there are pending sells (portfolio-level priority) ---
    def confirm_trade_entry(
        self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, **kwargs
    ) -> bool:
        """
        If ANY open trade exists for a pair listed in 'sell', we skip new buys this tick.
        (Best-effort global 'sell-first-then-buy' behavior.)
        """
        _, sell_list = self._fetch_signals()
        try:
            if self.dp:
                open_trades = self.dp.get_open_trades()
                for t in open_trades:
                    if t.pair.upper() in sell_list:
                        return False
        except Exception:
            # If dp is unavailable, don't block buys
            pass
        return True

    # --- Equal-share stake across current buy targets ---
    def custom_stake_amount(
        self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, **kwargs
    ) -> float:
        """
        Split available stake equally among pairs currently in the 'buy' list and not in 'sell'.
        """
        buy_list, sell_list = self._fetch_signals()
        # Candidate buy targets this tick
        targets = [p for p in buy_list if p not in sell_list]

        if not targets or pair.upper() not in targets:
            return proposed_stake  # no change

        try:
            # Best available API to get free stake:
            available = self.wallets.get_available_stake_amount()  # Freqtrade 2023+ API
        except Exception:
            try:
                available = self.wallets.get_total_stake_amount()  # fallback
            except Exception:
                available = proposed_stake

        # Divide equally. Keep a reasonable floor.
        n = max(len(targets), 1)
        stake = max(available / n, 0.0)

        # Respect min stake if exchange enforces it
        min_stake = getattr(self, "min_stake", 0) or 0
        if stake < min_stake:
            stake = max(min_stake, proposed_stake)

        return float(stake)