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.
Timeframe
1m
Direction
Long Only
Stoploss
-99.0%
Trailing Stop
No
ROI
0m: 1000.0%
Interface Version
N/A
Startup Candles
1
Indicators
0
freqtrade/freqtrade-strategies
freqtrade/freqtrade-strategies
this strategy is based around the idea of generating a lot of potentatils buys and make tiny profits on each trade
freqtrade/freqtrade-strategies
this strategy is based around the idea of generating a lot of potentatils buys and make tiny profits on each trade
# 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)