# source: https://raw.githubusercontent.com/jplirani/ow-dashboard/c1ca848fc7e0247b6b31f8abd70c7f2adfbe832d/SignalExtrategy.py # user_data/strategies/Github_jplirani_ow_dashboard__SignalExtrategy__20250823_202655.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 Github_jplirani_ow_dashboard__SignalExtrategy__20250823_202655(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"Github_jplirani_ow_dashboard__SignalExtrategy__20250823_202655: 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)