# source: https://raw.githubusercontent.com/Giansn/SYGNIF/74c8c15c0c291d935144d3e389df74f17f175702/SygnifStrategy.py """ SygnifStrategy - NFI-Enhanced Trading Bot with AI Sentiment Layer Architecture (inspired by NostalgiaForInfinityX7): - Multi-timeframe analysis: 5m base + 5m/15m/1h/4h/1d informative - BTC correlation: BTC/USDT indicators merged into all pairs - **Nautilus sidecar (optional):** ``user_data/btc_specialist_data/nautilus_strategy_signal.json`` (written by Docker ``nautilus-research`` bundle) → columns ``nautilus_bias`` / ``nautilus_signal_score`` → small TA-score nudge - NFI-style indicators: RSI_3/14, Aroon, StochRSI, CMF, CCI, ROC, BB, EMA, Williams %R - Global protections: Multi-TF cascade prevents buying during crashes - Sentiment layer: ambiguous TA zone → **local finance-agent HTTP** (`/sygnif/sentiment`) when `SYGNIF_USE_LOCAL_FINANCE_AGENT_HTTP=1` or `SYGNIF_SENTIMENT_HTTP_URL` is set — **rule-based** finance-agent expert score (-20..+20), no Haiku. If HTTP is not configured, optional **Cursor Cloud** / **Anthropic** paths remain (not used when HTTP URL is set). - NFI-style exit logic: Profit-tiered RSI exits + overbought signals + doom stoploss Cost: With `SYGNIF_USE_LOCAL_FINANCE_AGENT_HTTP=1`, sentiment is **non-LLM** on the node. See `SENTIMENT_BACKEND` only if you deliberately disable HTTP and use cloud/Anthropic. **Backend switch:** Set `SYGNIF_STRATEGY_BACKEND=ms2` (or `marketstrategy2`) so the Freqtrade class `SygnifStrategy` inherits **MarketStrategy2** while keeping `config["strategy"] = "SygnifStrategy"`. Default = **SygnifSentiment** stack (**Github_Giansn_SYGNIF__SygnifStrategy__20260428_213520**). """ import logging import os import json import re import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime import importlib.util from pathlib import Path from typing import Optional import feedparser import requests import numpy as np import pandas as pd try: from trade_overseer.event_log import EventLog from trade_overseer.risk_manager import RiskManager, RiskEngineConfig _HAS_OVERSEER = True except ImportError: _HAS_OVERSEER = False import pandas_ta as pta import talib.abstract as ta from freqtrade.strategy import IStrategy, merge_informative_pair from freqtrade.persistence import Trade from pandas import DataFrame from cursor_cloud_completion import cursor_cloud_completion from sentiment_constants import ( FINANCE_AGENT_SENTIMENT_INSTRUCTIONS, sentiment_tag_score_abs, tag_bypasses_premium_reserve, ) import btc_trend_regime logger = logging.getLogger(__name__) _sentiment_http_mod = None _mso_mod = None _adx_cdl_mod = None _vsd_mod = None _smc_mod = None _ml_mod = None def _get_market_sessions_orb(): """Lazy-load sibling ``market_sessions_orb`` (same pattern as sentiment_http_client).""" global _mso_mod if _mso_mod is None: mod_path = Path(__file__).resolve().parent / "market_sessions_orb.py" spec = importlib.util.spec_from_file_location("_sygnif_market_sessions_orb", mod_path) if spec is None or spec.loader is None: raise ImportError(f"cannot load market_sessions_orb from {mod_path}") _mso_mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(_mso_mod) return _mso_mod def _lazy_load_sibling(mod_name: str, cache_attr: str): """Generic lazy-loader for sibling strategy modules.""" g = globals() if g.get(cache_attr) is not None: return g[cache_attr] mod_path = Path(__file__).resolve().parent / f"{mod_name}.py" spec = importlib.util.spec_from_file_location(f"_sygnif_{mod_name}", mod_path) if spec is None or spec.loader is None: raise ImportError(f"cannot load {mod_name} from {mod_path}") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) g[cache_attr] = mod return mod def _get_adx_cdl(): return _lazy_load_sibling("adx_candlestick", "_adx_cdl_mod") def _get_vsd(): return _lazy_load_sibling("volume_sd_zones", "_vsd_mod") def _get_smc(): return _lazy_load_sibling("smc_indicators", "_smc_mod") def _get_ml(): return _lazy_load_sibling("ml_signal_ensemble", "_ml_mod") def _get_sentiment_http_client(): """Load sibling sentiment_http_client.py by path (avoids ModuleNotFoundError in Freqtrade workers).""" global _sentiment_http_mod if _sentiment_http_mod is None: mod_path = Path(__file__).resolve().parent / "sentiment_http_client.py" spec = importlib.util.spec_from_file_location("_sygnif_sentiment_http_client", mod_path) if spec is None or spec.loader is None: raise ImportError(f"cannot load sentiment_http_client from {mod_path}") _sentiment_http_mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(_sentiment_http_mod) return _sentiment_http_mod def _sygnif_strategy_backend() -> str: """Return ``ms2`` or ``sygnif`` from ``SYGNIF_STRATEGY_BACKEND`` (import-time).""" v = os.environ.get("SYGNIF_STRATEGY_BACKEND", "sygnif").strip().lower() if v in ("ms2", "marketstrategy2", "market_strategy_2"): return "ms2" return "sygnif" # --------------------------------------------------------------------------- # Sygnif Sentiment Layer # --------------------------------------------------------------------------- class SygnifSentiment: """Sentiment via local finance-agent HTTP (rule-based expert), or when HTTP is unset, Cursor Cloud Agent or Anthropic (legacy).""" def __init__(self): # auto | cursor_cloud | anthropic — auto prefers Cursor when CURSOR_* are set self._sentiment_backend = os.environ.get("SENTIMENT_BACKEND", "auto").strip().lower() self.api_key = os.environ.get("ANTHROPIC_API_KEY", "") self.model = "claude-haiku-4-5-20251001" self.base_url = "https://api.anthropic.com/v1/messages" self._cache: dict[str, tuple[float, float]] = {} self._news_cache: dict[str, tuple[float, list[str]]] = {} self.cache_ttl = 900 # 15 min cache (shorter for 5m TF) self.news_cache_ttl = 600 # 10 min news cache self.daily_calls = 0 self.daily_limit = 50 # Higher limit for 5m TF self._last_reset = datetime.now().date() # Monitoring counters self.non_zero_calls = 0 # Calls that returned a non-zero score self.parse_errors = 0 # JSON/regex parse failures self.api_errors = 0 # HTTP/network errors # Connection pooling — keep-alive for faster repeat calls self._session = requests.Session() # Circuit breaker — pause sentiment after N consecutive failures self._consecutive_failures = 0 self._circuit_open_until = 0.0 self._circuit_threshold = 5 # failures before opening self._circuit_cooldown = 300 # 5 min pause when open def _use_cursor_cloud(self) -> bool: b = self._sentiment_backend if b in ("cursor", "cursor_cloud"): return True if b == "anthropic": return False # auto: match finance_agent — same CURSOR_* when both set if os.environ.get("CURSOR_API_KEY", "").strip() and os.environ.get( "CURSOR_AGENT_REPOSITORY", "" ).strip(): return True return False def _parse_sentiment_score(self, text: str) -> Optional[float]: """Extract {\"score\": ...} from model output.""" if not text: return None match = re.search(r"\{[^{}]*\"score\"[^{}]*\}", text, re.DOTALL) if not match: return None try: result = json.loads(match.group(0)) return max(-20.0, min(20.0, float(result["score"]))) except (json.JSONDecodeError, KeyError, TypeError, ValueError): return None def _call_api_with_retry(self, payload: dict, max_attempts: int = 3) -> Optional[requests.Response]: """POST to Anthropic API with exponential backoff and circuit breaker. Returns the Response object on success, or None on failure. Honors circuit breaker — returns None immediately if open. """ # Circuit breaker check now = time.time() if now < self._circuit_open_until: return None headers = { "x-api-key": self.api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", } for attempt in range(max_attempts): try: resp = self._session.post( self.base_url, headers=headers, json=payload, timeout=20 ) # 429 rate limit — long backoff if resp.status_code == 429: wait = 5 * (2 ** attempt) # 5s, 10s, 20s logger.warning(f"Sentiment 429 rate limit, waiting {wait}s") time.sleep(wait) continue # 5xx server error — short backoff if 500 <= resp.status_code < 600: wait = 2 ** attempt # 1s, 2s, 4s logger.warning(f"Sentiment {resp.status_code}, retrying in {wait}s") time.sleep(wait) continue # 4xx other than 429 — fail fast, no retry if 400 <= resp.status_code < 500: self._consecutive_failures += 1 self._maybe_trip_breaker() return resp # Success self._consecutive_failures = 0 return resp except (requests.Timeout, requests.ConnectionError) as e: wait = 2 ** attempt logger.warning(f"Sentiment {type(e).__name__}, retrying in {wait}s") time.sleep(wait) except Exception as e: logger.error(f"Sentiment unexpected error: {e}") break # All retries exhausted self._consecutive_failures += 1 self._maybe_trip_breaker() return None def _maybe_trip_breaker(self): """Open the circuit breaker if too many consecutive failures.""" if self._consecutive_failures >= self._circuit_threshold: self._circuit_open_until = time.time() + self._circuit_cooldown logger.warning( f"Sentiment circuit breaker OPEN for {self._circuit_cooldown}s " f"after {self._consecutive_failures} consecutive failures" ) self._consecutive_failures = 0 # reset after opening def _reset_daily_counter(self): today = datetime.now().date() if today > self._last_reset: logger.info( f"Sentiment daily stats: calls={self.daily_calls}, " f"non_zero={self.non_zero_calls}, " f"parse_errors={self.parse_errors}, " f"api_errors={self.api_errors}" ) self.daily_calls = 0 self.non_zero_calls = 0 self.parse_errors = 0 self.api_errors = 0 self._last_reset = today def _get_cached(self, token: str) -> Optional[float]: if token in self._cache: ts, score = self._cache[token] if time.time() - ts < self.cache_ttl: return score return None def _fetch_rss(self, feed_url: str, token: str) -> list[str]: """Fetch headlines from a single RSS feed. Returns top 3 headlines regardless of token match — Claude decides relevance. Token-matching headlines are prioritized, then filled with general crypto news.""" try: feed = feedparser.parse(feed_url) token_matches = [] general = [] for entry in feed.entries[:5]: title = entry.get("title", "") if not title: continue if token.upper() in title.upper(): token_matches.append(title) else: general.append(title) return (token_matches + general)[:3] except Exception as e: logger.warning(f"Feed error {feed_url}: {e}") return [] def _fetch_reddit(self, token: str) -> list[str]: """Fetch top posts from r/CryptoCurrency mentioning the token.""" try: url = f"https://www.reddit.com/r/CryptoCurrency/search.json?q={token}&sort=new&limit=5&restrict_sr=1" resp = requests.get(url, headers={"User-Agent": "sygnif/1.0"}, timeout=5) if resp.ok: posts = resp.json().get("data", {}).get("children", []) return [p.get("data", {}).get("title", "") for p in posts[:3] if p.get("data", {}).get("title")] except Exception as e: logger.warning(f"Reddit fetch error for {token}: {e}") return [] def _fetch_gdelt(self, token: str) -> list[str]: """Fetch headlines from GDELT API.""" try: gdelt_url = ( f"https://api.gdeltproject.org/api/v2/doc/doc" f"?query={token}%20crypto&mode=artlist&maxrecords=5&format=json" ) resp = requests.get(gdelt_url, timeout=5) if resp.ok: return [art.get("title", "") for art in resp.json().get("articles", [])[:3]] except Exception: pass return [] def fetch_news(self, token: str, max_items: int = 5) -> list[str]: """Fetch recent crypto news from free RSS feeds (parallel, cached).""" if token in self._news_cache: ts, cached_headlines = self._news_cache[token] if time.time() - ts < self.news_cache_ttl: return cached_headlines feeds = [ f"https://cryptopanic.com/news/{token.lower()}/rss/", "https://cointelegraph.com/rss", "https://www.coindesk.com/arc/outboundfeeds/rss/", ] headlines = [] with ThreadPoolExecutor(max_workers=5) as pool: rss_futures = {pool.submit(self._fetch_rss, url, token): url for url in feeds} gdelt_future = pool.submit(self._fetch_gdelt, token) reddit_future = pool.submit(self._fetch_reddit, token) for future in as_completed(rss_futures): headlines.extend(future.result()) headlines.extend(gdelt_future.result()) headlines.extend(reddit_future.result()) # Dedupe while preserving order seen = set() deduped = [] for h in headlines: if h and h not in seen: seen.add(h) deduped.append(h) result = deduped[:max_items] self._news_cache[token] = (time.time(), result) return result def analyze_sentiment( self, token: str, current_price: float, ta_score: float, headlines: list[str], ) -> float: """ Prefer finance-agent HTTP (rule-based expert). Optionally Cursor or Anthropic only when no sentiment HTTP URL is configured. """ self._reset_daily_counter() cached = self._get_cached(token) if cached is not None: logger.info(f"Sentiment (cached) for {token}: {cached}") return cached http_url = os.environ.get("SYGNIF_SENTIMENT_HTTP_URL", "").strip() if not http_url and os.environ.get( "SYGNIF_USE_LOCAL_FINANCE_AGENT_HTTP", "" ).strip().lower() in ("1", "true", "yes", "on"): http_url = "http://127.0.0.1:8091/sygnif/sentiment" if http_url: ok, sc, err = _get_sentiment_http_client().post_sygnif_sentiment( http_url, token, current_price, ta_score, headlines ) if ok and sc is not None: self.daily_calls += 1 self._cache[token] = (time.time(), float(sc)) if sc != 0: self.non_zero_calls += 1 logger.info("Sentiment HTTP (%s) for %s: %s", http_url, token, sc) return float(sc) logger.warning("SYGNIF_SENTIMENT_HTTP failed (%s): %s", token, err) return None if self.daily_calls >= self.daily_limit: logger.warning("Sentiment daily limit reached, returning neutral") return 0.0 use_cursor = self._use_cursor_cloud() if not use_cursor and not self.api_key: logger.warning( "Skipping sentiment: set CURSOR_API_KEY + CURSOR_AGENT_REPOSITORY (Cursor) " "and/or ANTHROPIC_API_KEY (Haiku fallback)" ) return 0.0 news_text = "\n".join(f"- {h}" for h in headlines) if headlines else "No recent headlines available." prompt = f"""{FINANCE_AGENT_SENTIMENT_INSTRUCTIONS} Assess sentiment for trading {token}. Current price: ${current_price:.4f} Technical analysis score: {ta_score:.0f}/100 (50 = neutral, >60 = bullish, <40 = bearish) Recent crypto headlines (mix of {token}-specific and general market): {news_text} Provide a sentiment adjustment score combining: 1. Any {token}-specific news (most weight if present) 2. General crypto market mood from the headlines 3. Macro/regulatory context implied by the news Rules: - Score between -20 (strongly bearish) and +20 (strongly bullish) - 0 = no edge / pure noise / contradictory signals - It's OK to give a small score (±3 to ±8) based purely on general market mood when no {token}-specific news exists - Reserve ±15 to ±20 for major specific events (regulatory action, listings, hacks, partnerships) - If the headlines are pure noise or completely off-topic, return 0 Respond with ONLY a JSON object: {{"score": , "reason": ""}}""" self.daily_calls += 1 if use_cursor: text = cursor_cloud_completion(prompt, label="Sygnif sentiment") if text is None: self.api_errors += 1 self._consecutive_failures += 1 self._maybe_trip_breaker() logger.error(f"Cursor Cloud sentiment for {token}: failed or empty") # auto-fallback to Anthropic if configured if self._sentiment_backend == "auto" and self.api_key: logger.info(f"Falling back to Anthropic Haiku for {token}") use_cursor = False else: return None if use_cursor: score = self._parse_sentiment_score(text) if score is None: self.parse_errors += 1 logger.error(f"Cursor sentiment: no JSON in response: {text[:200] if text else ''}") return None reason = "" try: m = re.search(r"\{[^{}]*\"score\"[^{}]*\}", text, re.DOTALL) if m: reason = json.loads(m.group(0)).get("reason", "") or "" except Exception: pass if score != 0: self.non_zero_calls += 1 logger.info( f"Cursor sentiment for {token}: {score} — {reason} " f"[stats: {self.daily_calls}c {self.non_zero_calls}nz {self.parse_errors}pe {self.api_errors}ae]" ) self._cache[token] = (time.time(), score) self._consecutive_failures = 0 return float(score) # Anthropic Haiku path payload = { "model": self.model, "max_tokens": 100, "messages": [{"role": "user", "content": prompt}], } resp = self._call_api_with_retry(payload) if resp is None: self.api_errors += 1 logger.error(f"Claude sentiment for {token}: all retries failed (or circuit open)") return None if not resp.ok: self.api_errors += 1 logger.error(f"Claude API error: {resp.status_code} {resp.text[:200]}") return None try: data = resp.json() text = data["content"][0]["text"] score = self._parse_sentiment_score(text) if score is None: self.parse_errors += 1 logger.error(f"Claude sentiment: no JSON object found in response: {text[:200]}") return None reason = "" try: m = re.search(r"\{[^{}]*\"score\"[^{}]*\}", text, re.DOTALL) if m: reason = json.loads(m.group(0)).get("reason", "") or "" except Exception: pass if score != 0: self.non_zero_calls += 1 logger.info( f"Claude sentiment for {token}: {score} — {reason} " f"[stats: {self.daily_calls}c {self.non_zero_calls}nz {self.parse_errors}pe {self.api_errors}ae]" ) self._cache[token] = (time.time(), score) return score except Exception as e: self.parse_errors += 1 logger.error(f"Claude sentiment parse error: {e}") return None # --------------------------------------------------------------------------- # Strategy (default implementation — class SygnifStrategy is bound below) # --------------------------------------------------------------------------- class Github_Giansn_SYGNIF__SygnifStrategy__20260428_213520(IStrategy): """ Sygnif — NFI-Enhanced Freqtrade strategy with AI sentiment analysis. Based on NostalgiaForInfinityX7 patterns: - Multi-timeframe indicators (5m + 15m/1h/4h/1d) - BTC correlation - Global protections cascade - Profit-tiered RSI exits - Claude sentiment for ambiguous signals """ INTERFACE_VERSION = 3 can_short = False # Overridden to True in __init__ when futures mode # --- Core settings (NFI-style) --- stoploss = -0.20 # Base SL (overridden per-trade by custom_stoploss) trailing_stop = False use_custom_stoploss = True timeframe = "5m" info_timeframes = ["15m", "1h", "4h", "1d"] process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True startup_candle_count: int = 400 # Minimal ROI — let custom_exit handle exits minimal_roi = {"0": 100} # Protections — lock pair after repeated stoploss hits protections = [ {"method": "StoplossGuard", "lookback_period_candles": 12, "trade_limit": 2, "stop_duration_candles": 48, "only_per_pair": True}, {"method": "CooldownPeriod", "stop_duration_candles": 2}, ] # --- Thresholds --- stop_threshold_doom_spot = 0.20 # -20% doom stoploss (spot) stop_threshold_doom_futures = 0.20 # -20% doom stoploss (futures, divided by leverage) soft_sl_ratio_spot = 0.60 # soft SL at 60% of doom = -12% P&L (spot) soft_sl_ratio_futures = 0.60 # soft SL at 60% of doom = -12% P&L (futures, tighter due to margin/funding cost) # --- Leverage --- futures_mode_leverage = 3.0 futures_mode_leverage_majors = 5.0 # BTC, ETH, SOL, XRP — slow movers, higher leverage ok # Major pairs eligible for 5x major_pairs = ["BTC/USDT", "ETH/USDT", "SOL/USDT", "XRP/USDT"] # Sentiment thresholds sentiment_threshold_buy = 55.0 # Minimum |expert sentiment| for sygnif_s* / sygnif_short_s* entries (populate + confirm). sentiment_entry_min_abs = 3 sentiment_threshold_sell = 40.0 # Movers tracking — top gainers/losers refreshed every 4h _movers_pairs: list[str] = [] _movers_last_update: float = 0.0 _movers_refresh_secs: int = 14400 # 4h # New pairs tracking — externally sourced (sentiment scanner, new listings, etc) _new_pairs: list[str] = [] _new_pairs_last_update: float = 0.0 _new_pairs_refresh_secs: int = 1800 # 30 min (newer signals = more frequent refresh) _new_pairs_path = "user_data/new_pairs.json" # Doom cooldown — per-pair lockout after stoploss hit _doom_cooldown: dict[str, float] = {} doom_cooldown_secs = 14400 # 4h _doom_loss_count: dict[str, list[float]] = {} # consecutive loss tracking # Slot caps per entry type (prevent one type hogging all slots) max_slots_strong = 6 # strong_ta entries (TA >= strong_ta_min_score) max_slots_strong_short = 6 # strong_ta_short (futures) — mirrors long cap max_slots_swing = 4 # swing_failure, sygnif_swing (+ fa_/claude_ legacy aliases) _swing_tags = frozenset({ "swing_failure", "swing_failure_short", "sygnif_swing", "sygnif_swing_short", "fa_swing", "fa_swing_short", "claude_swing", "claude_swing_short", }) # Hybrid swing: EMA swing exit first, then Williams/RSI stack (excludes standalone swing_failure*) _swing_hybrid_long = frozenset({"sygnif_swing", "fa_swing", "claude_swing"}) _swing_hybrid_short = frozenset({"sygnif_swing_short", "fa_swing_short", "claude_swing_short"}) # --- Runtime tunables (defaults; overridden by user_data/strategy_adaptation.json) --- strong_ta_min_score = 65 strong_ta_short_max_score = 25 claude_long_score_low = 40 claude_long_score_high = 64 claude_short_score_low = 30 claude_short_score_high = 60 vol_strong_mult = 1.2 # Failure swing (Heavy91-style) — overridden by strategy_adaptation.json sf_lookback_bars = 48 sf_vol_filter_min = 0.03 sf_sl_base = 0.02 sf_sl_vol_scale = 0.02 sf_tp_vol_scale = 0.05 sf_ta_split = 50.0 # Session ORB (BTC/ETH, 5m) — ``market_sessions_orb.attach_orb_columns``; toggles via adaptation orb_entry_enabled = 0 max_slots_orb = 2 max_slots_btc_trend = 2 # SYGNIF_PROFILE=btc_trend — tag btc_trend_long orb_range_minutes = 30 orb_min_range_pct = 0.05 _adaptation_last_load: float = 0.0 _adaptation_refresh_secs: int = 60 # Premium tag reservation: non-premium entries are capped at # premium_nonreserved_max open trades. Remaining slots (max_open_trades - # premium_nonreserved_max) are reserved for tags in PREMIUM_TAGS only. # Also see ``tag_bypasses_premium_reserve`` (ORB, sygnif_swing, |sentiment|>=4). PREMIUM_TAGS = frozenset( {"sygnif_s-5", "sygnif_swing_short", "claude_s-5", "claude_swing_short"} ) premium_nonreserved_max = 10 # non-premium cap (used with max_open_trades=12) # Subclasses (e.g. ``BTC_Strategy_0_1``) may set tags here so futures # ``MIN_ACTIVE_VOLUME_PAIRS`` + per-pair volume gate do not block **all** # entries on a **BTC-only** (or very short) whitelist. _tags_bypass_volume_regime: frozenset = frozenset() # Claude layer sentiment = SygnifSentiment() # --- NT risk-based sizing (Lesson 2) --- RISK_PCT = 0.02 # 2% of equity risked per trade # --- DCA scale-in (Lesson 4) --- DCA_ELIGIBLE_TAGS = frozenset( {"sygnif_s-2", "sygnif_s-5", "claude_s-2", "claude_s-5"} ) DCA_DRAWDOWN_STEP = -0.03 DCA_MAX_ENTRIES = 1 DCA_SCALE_FACTOR = 0.5 # --- Volume regime gate (Lesson 5) --- MIN_ACTIVE_VOLUME_PAIRS = 3 _active_volume_pairs: int = 999 # default high so gate is open until first scan # ------------------------------------------------------------------------- # Enable shorts dynamically for futures mode # ------------------------------------------------------------------------- def bot_start(self, **kwargs) -> None: if self.config.get("trading_mode", "") == "futures": self.can_short = True self._load_doom_cooldown() self._refresh_movers() self._refresh_new_pairs() self._refresh_strategy_adaptation(force=True) # NT Lesson 2+3: risk engine + event log self._risk_manager = None self._event_log = None self._last_sl_tier: dict[int, str] = {} if _HAS_OVERSEER: self._risk_manager = RiskManager(RiskEngineConfig( ratchet_tiers=( (0.10, 0.015), (0.05, 0.02), ), )) instance = "freqtrade-futures" if self.config.get("trading_mode", "") == "futures" else "freqtrade" self._event_log = EventLog(instance=instance) # Nautilus sidecar JSON cache (mtime-invalidated) self._nautilus_signal_mtime: float = -1.0 self._nautilus_signal_doc: Optional[dict] = None def _nautilus_sidecar_json_path(self) -> Path: env = os.environ.get("NAUTILUS_SIGNAL_JSON", "").strip() if env: return Path(env).expanduser() return Path(__file__).resolve().parent.parent / "btc_specialist_data" / "nautilus_strategy_signal.json" def _load_nautilus_sidecar_signal(self) -> Optional[dict]: p = self._nautilus_sidecar_json_path() try: st = p.stat().st_mtime except OSError: return None if self._nautilus_signal_doc is not None and st == self._nautilus_signal_mtime: return self._nautilus_signal_doc try: doc = json.loads(p.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): self._nautilus_signal_doc = None self._nautilus_signal_mtime = st return None if not isinstance(doc, dict): self._nautilus_signal_doc = None self._nautilus_signal_mtime = st return None self._nautilus_signal_doc = doc self._nautilus_signal_mtime = st return doc def _attach_nautilus_sidecar_columns(self, df: DataFrame, metadata: dict) -> None: """Broadcast Nautilus container sidecar hint onto the dataframe (see ``nautilus_sidecar_strategy.py``).""" if os.environ.get("SYGNIF_NAUTILUS_SIGNAL", "1").lower() in ("0", "false", "no", "off"): df.loc[:, "nautilus_bias"] = "off" df.loc[:, "nautilus_signal_score"] = 0.0 return doc = self._load_nautilus_sidecar_signal() if not doc: df.loc[:, "nautilus_bias"] = "neutral" df.loc[:, "nautilus_signal_score"] = 0.0 return bias = str(doc.get("bias") or "neutral").lower() if bias not in ("long", "short", "neutral"): bias = "neutral" bonus = 3.0 if bias == "long" else (-3.0 if bias == "short" else 0.0) df.loc[:, "nautilus_bias"] = bias df.loc[:, "nautilus_signal_score"] = float(bonus) try: logger.debug( "[%s] nautilus sidecar: bias=%s score_delta=%s", metadata.get("pair", ""), bias, bonus, ) except Exception: pass def _refresh_strategy_adaptation(self, force: bool = False) -> None: """Load bounded overrides from user_data/strategy_adaptation.json (hot-reload).""" now = time.time() if not force and (now - self._adaptation_last_load) < self._adaptation_refresh_secs: return self._adaptation_last_load = now try: ud = Path(__file__).resolve().parent.parent mod_path = ud / "strategy_adaptation.py" json_path = ud / "strategy_adaptation.json" spec = importlib.util.spec_from_file_location("_sygnif_adapt", mod_path) if spec is None or spec.loader is None or not mod_path.is_file(): return mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) ovr = mod.load_adaptation_file(json_path) mod.apply_defaults_and_overrides(self, ovr) except Exception as e: logger.warning("strategy_adaptation: %s", e) def bot_loop_start(self, current_time=None, **kwargs) -> None: """Inject movers and externally-sourced new pairs into active whitelist.""" self._refresh_strategy_adaptation(force=False) self._refresh_movers() self._refresh_new_pairs() if not self.dp: return current_wl = self.dp.current_whitelist() for pair in self._movers_pairs: if pair not in current_wl: current_wl.append(pair) logger.info(f"Mover {pair} added to whitelist") for pair in self._new_pairs: if pair not in current_wl: current_wl.append(pair) logger.info(f"New pair {pair} added to whitelist") # NT Lesson 5: scan volume regime for futures if self.config.get("trading_mode", "") == "futures": count = 0 for p in current_wl: try: df, _ = self.dp.get_analyzed_dataframe(p, self.timeframe) if len(df) > 0 and self._futures_volume_gate_passes(df): count += 1 except Exception: pass self._active_volume_pairs = count # ------------------------------------------------------------------------- # New pairs integration — externally sourced (sentiment scanner, listings) # ------------------------------------------------------------------------- def _refresh_new_pairs(self): """Read user_data/new_pairs.json and inject pairs into whitelist. Expected file format (mirrors movers_pairlist.json): { "exchange": {"pair_whitelist": ["XRP/USDT", "DOGE/USDT", ...]}, "_meta": {"source": "finance-agent", "updated": "2026-04-06T..."} } Pairs in this file get the same whitelist injection as movers, bypassing the static VolumePairList. Use this for: - Pairs identified by external scanners (finance-agent) - Recently listed coins below the AgeFilter threshold - Manual pair additions without editing config """ now = time.time() if now - self._new_pairs_last_update < self._new_pairs_refresh_secs and self._new_pairs: return try: base = Path(__file__).resolve().parent while base != base.parent: candidate = base / self._new_pairs_path if candidate.exists(): break base = base.parent np_file = base / self._new_pairs_path if not np_file.exists(): # Silent — file is optional self._new_pairs_last_update = now return data = json.loads(np_file.read_text()) pairs = data.get("exchange", {}).get("pair_whitelist", []) # Filter for the right trading mode (spot vs futures syntax) is_futures = self.config.get("trading_mode", "") == "futures" valid = [] for p in pairs: if is_futures: # Futures pairs use BTC/USDT:USDT syntax if ":" not in p: p = f"{p}:USDT" valid.append(p) else: # Spot uses BTC/USDT (strip :USDT if present) if ":" in p: p = p.split(":")[0] valid.append(p) if valid: self._new_pairs = valid self._new_pairs_last_update = now meta = data.get("_meta", {}) logger.info( f"New pairs loaded ({len(valid)}): {valid[:5]}{'...' if len(valid)>5 else ''} " f"(source: {meta.get('source', 'unknown')})" ) except Exception as e: logger.warning(f"New pairs refresh failed: {e}") # ------------------------------------------------------------------------- # Doom cooldown persistence — survive restarts # ------------------------------------------------------------------------- _doom_cooldown_path = "user_data/doom_cooldown.json" def _load_doom_cooldown(self): try: with open(self._doom_cooldown_path) as f: data = json.load(f) now = time.time() self._doom_cooldown = { k: v for k, v in data.get("cooldowns", data).items() if now - v < self.doom_cooldown_secs } # Load consecutive loss tracking loss_counts = data.get("loss_counts", {}) self._doom_loss_count = { k: [t for t in v if now - t < 86400] for k, v in loss_counts.items() } except (FileNotFoundError, json.JSONDecodeError): self._doom_cooldown = {} self._doom_loss_count = {} def _save_doom_cooldown(self): try: with open(self._doom_cooldown_path, "w") as f: json.dump({ "cooldowns": self._doom_cooldown, "loss_counts": self._doom_loss_count, }, f) except OSError as e: logger.warning(f"Failed to save doom cooldown: {e}") # ------------------------------------------------------------------------- # Fetch top gainers/losers from movers file (refreshed every 4h) # ------------------------------------------------------------------------- def _refresh_movers(self): now = time.time() if now - self._movers_last_update < self._movers_refresh_secs and self._movers_pairs: return try: # Walk up from this file until we find user_data/ base = Path(__file__).resolve().parent while base != base.parent: candidate = base / "user_data" / "movers_pairlist.json" if candidate.exists(): break base = base.parent movers_file = base / "user_data" / "movers_pairlist.json" if not movers_file.exists(): logger.warning(f"Movers file not found: {movers_file}") return data = json.loads(movers_file.read_text()) pairs = data.get("exchange", {}).get("pair_whitelist", []) if pairs: self._movers_pairs = pairs self._movers_last_update = now meta = data.get("_meta", {}) logger.info( f"Movers loaded: gainers={meta.get('gainers', [])}, " f"losers={meta.get('losers', [])}" ) except Exception as e: logger.warning(f"Movers refresh failed: {e}") # ------------------------------------------------------------------------- # Informative pairs — BTC data + movers + new pairs # ------------------------------------------------------------------------- def informative_pairs(self): self._refresh_movers() self._refresh_new_pairs() is_futures = self.config.get("trading_mode", "") == "futures" btc_pair = "BTC/USDT:USDT" if is_futures else "BTC/USDT" pairs = [] # BTC correlation data for tf in self.info_timeframes: pairs.append((btc_pair, tf)) # All whitelist pairs need higher TF data for exits + crash protection whitelist = self.dp.current_whitelist() if self.dp else [] for pair in whitelist: for tf in self.info_timeframes: pairs.append((pair, tf)) # Movers (may not be in whitelist yet) for mover in self._movers_pairs: for tf in [self.timeframe] + self.info_timeframes: pairs.append((mover, tf)) # New pairs (externally sourced — sentiment scanner, listings) for new_pair in self._new_pairs: for tf in [self.timeframe] + self.info_timeframes: pairs.append((new_pair, tf)) # Dedupe pairs = list(dict.fromkeys(pairs)) return pairs # ------------------------------------------------------------------------- # BTC informative indicators # ------------------------------------------------------------------------- def btc_informative_indicators(self, btc_df: DataFrame, timeframe: str) -> DataFrame: btc_df["btc_RSI_3"] = pta.rsi(btc_df["close"], length=3) btc_df["btc_RSI_14"] = pta.rsi(btc_df["close"], length=14) _r = pta.ema(btc_df["close"], length=200) btc_df["btc_EMA_200"] = _r if _r is not None else np.nan btc_df["btc_change_pct"] = (btc_df["close"] - btc_df["open"]) / btc_df["open"] * 100.0 # Rename to avoid collision ignore_columns = ["date", "btc_RSI_3", "btc_RSI_14", "btc_EMA_200", "btc_change_pct"] btc_df.drop(columns=[c for c in btc_df.columns if c not in ignore_columns], inplace=True) return btc_df # ------------------------------------------------------------------------- # Informative timeframe indicators # ------------------------------------------------------------------------- def informative_indicators(self, df: DataFrame, timeframe: str) -> DataFrame: if len(df) < 2 or "close" not in df.columns: return df # RSI df["RSI_3"] = pta.rsi(df["close"], length=3) df["RSI_14"] = pta.rsi(df["close"], length=14) df["RSI_3_change_pct"] = df["RSI_3"].pct_change() * 100.0 # EMA df["EMA_12"] = pta.ema(df["close"], length=12) df["EMA_200"] = pta.ema(df["close"], length=200, ) # BB if len(df) >= 20: bbands = pta.bbands(df["close"], length=20) if isinstance(bbands, pd.DataFrame) and "BBL_20_2.0" in bbands.columns: df["BBL_20_2.0"] = bbands["BBL_20_2.0"] df["BBM_20_2.0"] = bbands["BBM_20_2.0"] df["BBU_20_2.0"] = bbands["BBU_20_2.0"] for col in ["BBL_20_2.0", "BBM_20_2.0", "BBU_20_2.0"]: if col not in df.columns: df[col] = np.nan # CMF df["CMF_20"] = pta.cmf(df["high"], df["low"], df["close"], df["volume"], length=20) # Aroon aroon = pta.aroon(df["high"], df["low"], length=14) if isinstance(aroon, pd.DataFrame) and "AROONU_14" in aroon.columns: df["AROONU_14"] = aroon["AROONU_14"] df["AROOND_14"] = aroon["AROOND_14"] for col in ["AROONU_14", "AROOND_14"]: if col not in df.columns: df[col] = np.nan # StochRSI stochrsi = pta.stochrsi(df["close"]) if isinstance(stochrsi, pd.DataFrame) and "STOCHRSIk_14_14_3_3" in stochrsi.columns: df["STOCHRSIk_14_14_3_3"] = stochrsi["STOCHRSIk_14_14_3_3"] if "STOCHRSIk_14_14_3_3" not in df.columns: df["STOCHRSIk_14_14_3_3"] = np.nan # CCI df["CCI_20"] = pta.cci(df["high"], df["low"], df["close"], length=20) # ROC df["ROC_9"] = pta.roc(df["close"], length=9) # Williams %R df["WILLR_14"] = pta.willr(df["high"], df["low"], df["close"], length=14) return df # ------------------------------------------------------------------------- # Populate indicators # ------------------------------------------------------------------------- def populate_indicators(self, df: DataFrame, metadata: dict) -> DataFrame: if len(df) < self.startup_candle_count: return df try: return self._populate_indicators_inner(df, metadata) except (ValueError, KeyError) as e: logger.warning(f"[{metadata.get('pair')}] Skipping indicators: {e}") return df def _populate_indicators_inner(self, df: DataFrame, metadata: dict) -> DataFrame: tik = time.perf_counter() # --- BTC informative (all timeframes) --- is_futures = self.config.get("trading_mode", "") == "futures" btc_pair = "BTC/USDT:USDT" if is_futures else "BTC/USDT" if metadata["pair"] != btc_pair: for tf in self.info_timeframes: btc_df = self.dp.get_pair_dataframe(btc_pair, tf) if len(btc_df) < 2: continue btc_df = self.btc_informative_indicators(btc_df, tf) df = merge_informative_pair(df, btc_df, self.timeframe, tf, ffill=True) drop_cols = [f"date_{tf}"] df.drop(columns=df.columns.intersection(drop_cols), inplace=True) # --- Informative timeframes --- for tf in self.info_timeframes: info_df = self.dp.get_pair_dataframe(metadata["pair"], tf) if len(info_df) < 2: continue info_df = self.informative_indicators(info_df, tf) df = merge_informative_pair(df, info_df, self.timeframe, tf, ffill=True) drop_cols = [f"date_{tf}", f"open_{tf}", f"high_{tf}", f"low_{tf}", f"close_{tf}", f"volume_{tf}"] df.drop(columns=df.columns.intersection(drop_cols), inplace=True) # --- Base 5m indicators (full NFI set) --- # RSI df["RSI_3"] = pta.rsi(df["close"], length=3) df["RSI_4"] = pta.rsi(df["close"], length=4) df["RSI_14"] = pta.rsi(df["close"], length=14) df["RSI_20"] = pta.rsi(df["close"], length=20) df["RSI_3_change_pct"] = df["RSI_3"].pct_change() * 100.0 df["RSI_14_change_pct"] = df["RSI_14"].pct_change() * 100.0 df["RSI_14_shift3"] = df["RSI_14"].shift(3) # EMA (full spectrum) df["EMA_3"] = pta.ema(df["close"], length=3) df["EMA_9"] = pta.ema(df["close"], length=9) df["EMA_12"] = pta.ema(df["close"], length=12) df["EMA_16"] = pta.ema(df["close"], length=16) df["EMA_20"] = pta.ema(df["close"], length=20) df["EMA_26"] = pta.ema(df["close"], length=26) df["EMA_50"] = pta.ema(df["close"], length=50) _r = pta.ema(df["close"], length=100) df["EMA_100"] = _r if _r is not None else np.nan _r = pta.ema(df["close"], length=200) df["EMA_200"] = _r if _r is not None else np.nan # SMA df["SMA_9"] = pta.sma(df["close"], length=9) df["SMA_16"] = pta.sma(df["close"], length=16) df["SMA_21"] = pta.sma(df["close"], length=21) df["SMA_30"] = pta.sma(df["close"], length=30) df["SMA_200"] = pta.sma(df["close"], length=200) # BB 20 - STD2 bbands_20 = pta.bbands(df["close"], length=20) if isinstance(bbands_20, pd.DataFrame) and "BBL_20_2.0" in bbands_20.columns: df["BBL_20_2.0"] = bbands_20["BBL_20_2.0"] df["BBM_20_2.0"] = bbands_20["BBM_20_2.0"] df["BBU_20_2.0"] = bbands_20["BBU_20_2.0"] df["BBB_20_2.0"] = bbands_20["BBB_20_2.0"] df["BBP_20_2.0"] = bbands_20["BBP_20_2.0"] for col in ["BBL_20_2.0", "BBM_20_2.0", "BBU_20_2.0", "BBB_20_2.0", "BBP_20_2.0"]: if col not in df.columns: df[col] = np.nan # BB 40 - STD2 try: bb40_upper, bb40_middle, bb40_lower = ta.BBANDS(df["close"], timeperiod=40, nbdevup=2.0, nbdevdn=2.0, matype=0) df["BBL_40_2.0"] = bb40_lower df["BBM_40_2.0"] = bb40_middle df["BBU_40_2.0"] = bb40_upper df["BBB_40_2.0"] = (bb40_upper - bb40_lower) / bb40_middle * 100.0 df["BBP_40_2.0"] = (df["close"] - bb40_lower) / (bb40_upper - bb40_lower) except Exception: for col in ["BBL_40_2.0", "BBM_40_2.0", "BBU_40_2.0", "BBB_40_2.0", "BBP_40_2.0"]: df[col] = np.nan # MFI df["MFI_14"] = pta.mfi(df["high"], df["low"], df["close"], df["volume"], length=14) # CMF df["CMF_20"] = pta.cmf(df["high"], df["low"], df["close"], df["volume"], length=20) # Williams %R df["WILLR_14"] = pta.willr(df["high"], df["low"], df["close"], length=14) df["WILLR_480"] = pta.willr(df["high"], df["low"], df["close"], length=480) # Aroon aroon = pta.aroon(df["high"], df["low"], length=14) if isinstance(aroon, pd.DataFrame): df["AROONU_14"] = aroon["AROONU_14"] df["AROOND_14"] = aroon["AROOND_14"] # StochRSI (k + d) stochrsi = pta.stochrsi(df["close"]) if isinstance(stochrsi, pd.DataFrame): df["STOCHRSIk_14_14_3_3"] = stochrsi["STOCHRSIk_14_14_3_3"] df["STOCHRSId_14_14_3_3"] = stochrsi["STOCHRSId_14_14_3_3"] # KST kst = pta.kst(df["close"]) if isinstance(kst, pd.DataFrame): df["KST_10_15_20_30_10_10_10_15"] = kst["KST_10_15_20_30_10_10_10_15"] df["KSTs_9"] = kst["KSTs_9"] # CCI df["CCI_20"] = pta.cci(df["high"], df["low"], df["close"], length=20) # ROC df["ROC_2"] = pta.roc(df["close"], length=2) df["ROC_9"] = pta.roc(df["close"], length=9) # OBV df["OBV"] = pta.obv(df["close"], df["volume"]) df["OBV_change_pct"] = df["OBV"].pct_change() * 100.0 # Candle stats df["change_pct"] = (df["close"] - df["open"]) / df["open"] * 100.0 df["close_delta"] = (df["close"] - df["close"].shift()).abs() df["close_max_6"] = df["close"].rolling(6).max() df["close_max_12"] = df["close"].rolling(12).max() df["close_max_48"] = df["close"].rolling(48).max() df["close_min_6"] = df["close"].rolling(6).min() df["close_min_12"] = df["close"].rolling(12).min() df["close_min_48"] = df["close"].rolling(48).min() df["volume_sma_25"] = pta.sma(df["volume"], length=25) df["ATR_14"] = pta.atr(df["high"], df["low"], df["close"], length=14) df["num_empty_288"] = (df["volume"] <= 0).rolling(window=288, min_periods=288).sum() # --- Handle NaN for merged columns --- for col in ["RSI_14_1h", "RSI_14_4h", "RSI_14_1d"]: if col in df.columns: df[col] = df[col].astype(np.float64).fillna(50.0) # --- Failure Swing (Stop Hunt) indicators --- lb = max(8, int(self.sf_lookback_bars)) df["sf_resistance"] = df["high"].shift(1).rolling(lb).max() df["sf_support"] = df["low"].shift(1).rolling(lb).min() # Stable level check: S/R unchanged for 2 bars (level is established) df["sf_resistance_stable"] = df["sf_resistance"] == df["sf_resistance"].shift(1) df["sf_support_stable"] = df["sf_support"] == df["sf_support"].shift(1) # EMA 120 for TP target df["EMA_120"] = pta.ema(df["close"], length=120) # Volatility filter: distance from EMA as fraction (blocks entries near EMA) df["sf_volatility"] = ((df["close"] - df["EMA_120"]).abs() / df["EMA_120"]) df["sf_vol_filter"] = df["sf_volatility"] > float(self.sf_vol_filter_min) # Long signal: wick below support but close back above + stable level df["sf_long"] = ( (df["low"] <= df["sf_support"]) & (df["close"] > df["sf_support"]) & df["sf_support_stable"] & df["sf_vol_filter"] ) # Short signal: wick above resistance but close back below + stable level df["sf_short"] = ( (df["high"] >= df["sf_resistance"]) & (df["close"] < df["sf_resistance"]) & df["sf_resistance_stable"] & df["sf_vol_filter"] ) # Dynamic SL/TP coefficients (Heavy91) df["sf_sl_pct"] = float(self.sf_sl_base) + df["sf_volatility"] * float(self.sf_sl_vol_scale) df["sf_tp_ema"] = df["EMA_120"] * ( 1 + df["sf_volatility"] * float(self.sf_tp_vol_scale) * np.sign(df["close"] - df["EMA_120"]) ) # --- Global protections (NFI-style) --- df["protections_long_global"] = self._calc_global_protections(df) df["protections_short_global"] = self._calc_global_protections_short(df) # --- Exchange downtime protection --- if self.dp.runmode.value in ("live", "dry_run"): df["live_data_ok"] = df["volume"].rolling(window=72, min_periods=72).min() > 0 # --- Session ORB (BTC/ETH spot+futures; optional, adaptation ``orb_entry_enabled``) --- if int(getattr(self, "orb_entry_enabled", 0)): try: tf = str(getattr(self, "timeframe", "5m") or "5m") tfm = int("".join(ch for ch in tf if ch.isdigit()) or "5") _mso = _get_market_sessions_orb() _mso.attach_orb_columns( df, metadata_pair=metadata.get("pair", ""), timeframe_minutes=tfm, orb_minutes=int(getattr(self, "orb_range_minutes", 30)), min_range_pct=float(getattr(self, "orb_min_range_pct", 0.05)), ) except Exception as e: logger.warning("[%s] ORB columns skipped: %s", metadata.get("pair"), e) # --- P1: ADX trend strength --- try: _get_adx_cdl().attach_adx(df, length=14) except Exception as e: logger.debug("[%s] ADX skipped: %s", metadata.get("pair"), e) # --- P2: Candlestick patterns (top 6) --- try: _get_adx_cdl().attach_candlestick_patterns(df) df["cdl_net_bullish"] = _get_adx_cdl().candlestick_bullish_score(df) except Exception as e: logger.debug("[%s] CDL patterns skipped: %s", metadata.get("pair"), e) # --- P3: Smart Money Concepts --- try: _get_smc().attach_smc_indicators(df, swing_length=10) except Exception as e: logger.debug("[%s] SMC skipped: %s", metadata.get("pair"), e) # --- P4: Volume S/D Zones (Heavy91 port) --- try: _get_vsd().attach_volume_sd_zones(df) except Exception as e: logger.debug("[%s] VSD zones skipped: %s", metadata.get("pair"), e) # --- P5: ML Signal Ensemble --- try: _get_ml().attach_ml_signal(df) except Exception as e: logger.debug("[%s] ML signal skipped: %s", metadata.get("pair"), e) self._attach_nautilus_sidecar_columns(df, metadata) tok = time.perf_counter() logger.debug(f"[{metadata['pair']}] populate_indicators took: {tok - tik:0.4f}s") return df # ------------------------------------------------------------------------- # Global protections — NFI-style multi-TF cascade # ------------------------------------------------------------------------- def _calc_global_protections(self, df: DataFrame) -> pd.Series: """ Prevent buying when multiple timeframes confirm a crash. Each clause is an OR triplet — entry only blocked when ALL conditions in a clause fail. """ prot = pd.Series(True, index=df.index) # 5m & 15m & 1h down move, higher TFs still not low enough if "RSI_3_15m" in df.columns and "RSI_3_1h" in df.columns: prot &= ( (df["RSI_3"] > 2.0) | (df["RSI_3_15m"] > 15.0) | (df["RSI_3_1h"] > 20.0) ) # 5m & 15m down, 1h & 4h still high prot &= ( (df["RSI_3"] > 3.0) | (df["RSI_3_15m"] > 10.0) | (df.get("RSI_14_1h", 50.0) < 40.0) ) # 5m & 1h down move, 4h still high if "RSI_3_1h" in df.columns and "RSI_14_4h" in df.columns: prot &= ( (df["RSI_3"] > 3.0) | (df["RSI_3_1h"] > 25.0) | (df["RSI_14_4h"] < 50.0) ) # 15m & 1h & 4h down move if "RSI_3_15m" in df.columns and "RSI_3_1h" in df.columns and "RSI_3_4h" in df.columns: prot &= ( (df["RSI_3_15m"] > 5.0) | (df["RSI_3_1h"] > 10.0) | (df["RSI_3_4h"] > 15.0) ) # 1h & 4h down move, 4h downtrend if "RSI_3_1h" in df.columns and "RSI_3_4h" in df.columns and "ROC_9_4h" in df.columns: prot &= ( (df["RSI_3_1h"] > 5.0) | (df["RSI_3_4h"] > 10.0) | (df["ROC_9_4h"] > -20.0) ) # 15m down, 15m & 4h still high (Aroon) if "RSI_3_15m" in df.columns and "AROONU_14_15m" in df.columns and "AROONU_14_4h" in df.columns: prot &= ( (df["RSI_3_15m"] > 5.0) | (df["AROONU_14_15m"] < 50.0) | (df["AROONU_14_4h"] < 60.0) ) # 5m down, 1h downtrend (CMF), 4h high if "CMF_20_1h" in df.columns and "AROONU_14_4h" in df.columns: prot &= ( (df["RSI_3"] > 5.0) | (df["CMF_20_1h"] > -0.25) | (df["AROONU_14_4h"] < 70.0) ) # 15m & 4h down, 1d downtrend if "RSI_3_15m" in df.columns and "RSI_3_4h" in df.columns and "ROC_9_1d" in df.columns: prot &= ( (df["RSI_3_15m"] > 5.0) | (df["RSI_3_4h"] > 10.0) | (df["ROC_9_1d"] > -40.0) ) # BTC crash protection (1h micro-RSI vs 4h washout — tightened vs 2026-04 dumps) if "btc_RSI_3_1h" in df.columns: prot &= ( (df["btc_RSI_3_1h"] > 18.0) | (df.get("btc_RSI_14_4h", 50.0) < 32.0) ) # BTC 1h dump + weak momentum: block fresh longs when the 1h candle is red, 1h RSI # is soft, and 4h has not yet washed out (catches slow bleeds RSI_3 alone misses). if "btc_change_pct_1h" in df.columns: ch1 = pd.to_numeric(df["btc_change_pct_1h"], errors="coerce") r1h = pd.to_numeric( df.get("btc_RSI_14_1h", pd.Series(50.0, index=df.index)), errors="coerce", ).fillna(50.0) r4h = pd.to_numeric( df.get("btc_RSI_14_4h", pd.Series(50.0, index=df.index)), errors="coerce", ).fillna(50.0) prot &= ( (ch1 > -0.85) | (r1h > 41.0) | (r4h < 35.0) ) return prot # ------------------------------------------------------------------------- # Global protections for SHORTS — NFI-style (inverse of long protections) # Prevent shorting when multiple timeframes confirm a pump / strong uptrend # ------------------------------------------------------------------------- def _calc_global_protections_short(self, df: DataFrame) -> pd.Series: prot = pd.Series(True, index=df.index) # 5m & 15m & 1h up move — don't short into a rally if "RSI_3_15m" in df.columns and "RSI_3_1h" in df.columns: prot &= ( (df["RSI_3"] < 90.0) | (df["RSI_3_15m"] < 75.0) | (df["RSI_3_1h"] < 75.0) ) prot &= ( (df["RSI_3"] < 95.0) | (df["RSI_3_15m"] < 85.0) | (df.get("RSI_14_1h", 50.0) > 60.0) ) # 5m & 1h up move, 4h still low — don't short early rally if "RSI_3_1h" in df.columns and "RSI_14_4h" in df.columns: prot &= ( (df["RSI_3"] < 95.0) | (df["RSI_3_1h"] < 75.0) | (df["RSI_14_4h"] > 50.0) ) # 15m & 1h & 4h up move if "RSI_3_15m" in df.columns and "RSI_3_1h" in df.columns and "RSI_3_4h" in df.columns: prot &= ( (df["RSI_3_15m"] < 90.0) | (df["RSI_3_1h"] < 85.0) | (df["RSI_3_4h"] < 80.0) ) # 1h & 4h up move, 4h uptrend if "RSI_3_1h" in df.columns and "RSI_3_4h" in df.columns and "ROC_9_4h" in df.columns: prot &= ( (df["RSI_3_1h"] < 90.0) | (df["RSI_3_4h"] < 85.0) | (df["ROC_9_4h"] < 20.0) ) # Aroon uptrend — don't short strong trend if "AROONU_14_15m" in df.columns and "AROONU_14_4h" in df.columns: prot &= ( (df["RSI_3_15m"] < 90.0) | (df["AROONU_14_15m"] < 75.0) | (df["AROONU_14_4h"] < 75.0) ) # BTC pump — don't short alts during BTC rally if "btc_RSI_3_1h" in df.columns: prot &= ( (df["btc_RSI_3_1h"] < 85.0) | (df.get("btc_RSI_14_4h", 50.0) > 70.0) ) # BTC structural uptrend filter — block ALL shorts when BTC 4h RSI > 60 # This is a hard structural filter, not just momentum-based if "btc_RSI_14_4h" in df.columns: prot &= df["btc_RSI_14_4h"].fillna(50.0) <= 60.0 return prot # ------------------------------------------------------------------------- # Leverage callback — NFI-style dynamic leverage # ------------------------------------------------------------------------- def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs ) -> float: # 5x for majors (BTC, ETH, SOL, XRP) if (pair.split(":")[0] if ":" in pair else pair) in self.major_pairs: tier_lev = self.futures_mode_leverage_majors else: # 3x default tier_lev = self.futures_mode_leverage # Shorts capped at 2x — unlimited upside risk in crypto if side == "short": tier_lev = min(tier_lev, 2.0) # Volatility cap: high ATR% → lower leverage try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(df) >= 14 and "ATR_14" in df.columns: atr_pct = (df["ATR_14"].iloc[-1] / df["close"].iloc[-1]) * 100 if atr_pct > 3.0: tier_lev = min(tier_lev, 2.0) elif atr_pct > 2.0: tier_lev = min(tier_lev, 3.0) except Exception: pass return min(tier_lev, max_leverage) # ------------------------------------------------------------------------- # Optional trailing TP/SL (Pine-style trail + fixed entry SL + post-TP lock; # trailing take-profit via pullback from extreme — Toobit-style callback). # Enable: SYGNIF_TRAILING_TPSL=1 — see .cursor/rules/trailing-tpsl.mdc # ------------------------------------------------------------------------- def _sygnif_trailing_tpsl_enabled(self) -> bool: v = os.environ.get("SYGNIF_TRAILING_TPSL", "").strip().lower() return v in ("1", "true", "yes", "on") def _sygnif_trailing_tpsl_float(self, key: str, default: float) -> float: raw = os.environ.get(key, "").strip() if not raw: return default try: return float(raw) except ValueError: return default def _sygnif_trailing_tpsl_params(self) -> dict: return { "trail_pct": max(1e-9, self._sygnif_trailing_tpsl_float("SYGNIF_TS_TRAIL_PCT", 0.07)), "fixed_sl_pct": max(1e-9, self._sygnif_trailing_tpsl_float("SYGNIF_TS_FIXED_SL_PCT", 0.02)), "tp_hit_pct": max(1e-9, self._sygnif_trailing_tpsl_float("SYGNIF_TS_TP_HIT_PCT", 0.02)), "post_tp_lock_pct": max(0.0, self._sygnif_trailing_tpsl_float("SYGNIF_TS_POST_TP_LOCK_PCT", 0.01)), "tp_callback_pct": max(1e-9, self._sygnif_trailing_tpsl_float("SYGNIF_TS_TP_CALLBACK_PCT", 0.01)), } def _stoploss_trailing_tpsl_combined( self, trade: Trade, current_rate: float, is_futures: bool, leverage: float, ) -> Optional[tuple[float, str]]: """Return (stoploss, tier) for combined trail + fixed SL (+ post-TP lock), or None.""" if not self._sygnif_trailing_tpsl_enabled(): return None enter_tag = trade.enter_tag or "" if enter_tag in self._swing_tags: return None entry = float(trade.open_rate) if entry <= 0 or current_rate <= 0: return None p = self._sygnif_trailing_tpsl_params() sl_doom = self.stop_threshold_doom_futures if is_futures else self.stop_threshold_doom_spot doom_long = -(sl_doom / leverage) if is_futures else -sl_doom doom_short = (sl_doom / leverage) if is_futures else sl_doom is_short = bool(getattr(trade, "is_short", False)) if is_short: mr = getattr(trade, "min_rate", None) trough = float(mr) if mr is not None and float(mr) > 0 else current_rate trough = min(trough, current_rate) post_tp = trough <= entry * (1.0 - p["tp_hit_pct"]) sl_trail = trough * (1.0 + p["trail_pct"]) sl_fixed = entry * (1.0 + p["fixed_sl_pct"]) if post_tp: sl_lock = entry * (1.0 - p["post_tp_lock_pct"]) effective_sl = min(sl_trail, sl_fixed, sl_lock) else: effective_sl = min(sl_trail, sl_fixed) r = (effective_sl / current_rate) - 1.0 if r <= 0: r = 1e-8 r = min(r, doom_short) return (r, "tsl_combined_short") # long xr = getattr(trade, "max_rate", None) peak = float(xr) if xr is not None and float(xr) > 0 else entry peak = max(peak, current_rate, entry) post_tp = peak >= entry * (1.0 + p["tp_hit_pct"]) sl_trail = peak * (1.0 - p["trail_pct"]) sl_fixed = entry * (1.0 - p["fixed_sl_pct"]) if post_tp: sl_lock = entry * (1.0 + p["post_tp_lock_pct"]) effective_sl = max(sl_trail, sl_fixed, sl_lock) else: effective_sl = max(sl_trail, sl_fixed) r = (effective_sl / current_rate) - 1.0 if r >= 0: r = -1e-8 r = max(r, doom_long) return (r, "tsl_combined_long") def _sygnif_exit_trailing_take_profit(self, trade: Trade, current_rate: float) -> Optional[str]: """Exit when price pulls back tp_callback_pct from extreme after tp_hit_pct move (long/short).""" if not self._sygnif_trailing_tpsl_enabled(): return None if (trade.enter_tag or "") in self._swing_tags: return None entry = float(trade.open_rate) if entry <= 0 or current_rate <= 0: return None p = self._sygnif_trailing_tpsl_params() cb = p["tp_callback_pct"] hit = p["tp_hit_pct"] if bool(getattr(trade, "is_short", False)): mr = getattr(trade, "min_rate", None) trough = float(mr) if mr is not None and float(mr) > 0 else current_rate trough = min(trough, current_rate) if trough > entry * (1.0 - hit): return None if current_rate >= trough * (1.0 + cb): return "exit_trailing_take_profit" return None peak = getattr(trade, "max_rate", None) peak_f = float(peak) if peak is not None and float(peak) > 0 else current_rate peak_f = max(peak_f, current_rate) if peak_f < entry * (1.0 + hit): return None if current_rate <= peak_f * (1.0 - cb): return "exit_trailing_take_profit" return None # ------------------------------------------------------------------------- # Custom stoploss — leverage-aware, placed on exchange # ------------------------------------------------------------------------- def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float: is_futures = self.config.get("trading_mode", "") == "futures" leverage = trade.leverage or 1.0 enter_tag = trade.enter_tag or "" # --- Swing failure trades: use their own dynamic SL (on-exchange) --- if enter_tag in self._swing_tags: try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(df) > 0 and "sf_sl_pct" in df.columns: sf_sl = df.iloc[-1].get("sf_sl_pct", 0.02) if current_profit >= 0.05: sl_val, tier = -0.02, "sf_ratchet_5pct" elif current_profit >= 0.02: sl_val, tier = -0.03, "sf_ratchet_2pct" else: sl_val, tier = -sf_sl, "sf_dynamic" self._emit_sl_tier(trade, pair, tier, sl_val, current_profit) return sl_val except Exception: pass self._emit_sl_tier(trade, pair, "sf_fallback", -0.03, current_profit) return -0.03 tsl = self._stoploss_trailing_tpsl_combined(trade, current_rate, is_futures, leverage) if tsl is not None: sl_val, tier = tsl self._emit_sl_tier(trade, pair, tier, sl_val, current_profit) return sl_val # --- Ratcheting trail: tighten SL as profit grows --- # +1% and +2% tiers removed (NT Lesson 1): they clipped winners at +0.70% avg # while indicator exits capture +7.60%. With risk-based sizing, doom costs # a fixed % of equity so the safety net is the sizing, not the ratchet. if current_profit >= 0.10: sl_val, tier = -0.015, "ratchet_10pct" elif current_profit >= 0.05: sl_val, tier = -0.02, "ratchet_5pct" else: # Base SL: fixed doom stoploss sl = self.stop_threshold_doom_futures if is_futures else self.stop_threshold_doom_spot if is_futures: sl_val = -(sl / leverage) else: sl_val = -sl tier = "doom" self._emit_sl_tier(trade, pair, tier, sl_val, current_profit) return sl_val def _emit_sl_tier(self, trade: Trade, pair: str, tier: str, sl_val: float, current_profit: float) -> None: """Emit order_updated event when the active SL tier changes (Lesson 3).""" if not hasattr(self, "_event_log") or self._event_log is None: return trade_id = getattr(trade, "id", None) if trade_id is None: return prev_tier = self._last_sl_tier.get(trade_id) if prev_tier == tier: return self._last_sl_tier[trade_id] = tier try: self._event_log.emit( "order_updated", instrument_id=pair, trade_id=trade_id, data={ "sl_tier": tier, "sl_value": sl_val, "current_profit": round(current_profit, 6), "leverage": trade.leverage or 1.0, "enter_tag": trade.enter_tag or "", }, ) except Exception: pass # ------------------------------------------------------------------------- # Calculate vectorized TA score (enhanced with NFI indicators) # ------------------------------------------------------------------------- def _calculate_ta_score_vectorized(self, df: DataFrame) -> pd.Series: score = pd.Series(50.0, index=df.index) # RSI_14 component (-15 to +15) rsi = df["RSI_14"].fillna(50.0) score += np.where(rsi < 30, 15, np.where(rsi < 40, 8, np.where(rsi > 70, -15, np.where(rsi > 60, -8, 0)))) # RSI_3 momentum (-10 to +10) rsi3 = df["RSI_3"].fillna(50.0) score += np.where(rsi3 < 10, 10, np.where(rsi3 < 20, 5, np.where(rsi3 > 90, -10, np.where(rsi3 > 80, -5, 0)))) # EMA crossover (-10 to +10) ema_bull = df["EMA_9"] > df["EMA_26"] ema_cross = ema_bull & (df["EMA_9"].shift(1) <= df["EMA_26"].shift(1)) score += np.where(ema_cross, 10, np.where(ema_bull, 7, -7)) # Bollinger (-8 to +8) if "BBL_20_2.0" in df.columns and "BBU_20_2.0" in df.columns: score += np.where(df["close"] <= df["BBL_20_2.0"], 8, np.where(df["close"] >= df["BBU_20_2.0"], -8, 0)) # Aroon (-8 to +8) if "AROONU_14" in df.columns and "AROOND_14" in df.columns: aroonu = df["AROONU_14"].fillna(50) aroond = df["AROOND_14"].fillna(50) score += np.where((aroonu > 80) & (aroond < 30), 8, np.where((aroond > 80) & (aroonu < 30), -8, 0)) # StochRSI (-5 to +5) if "STOCHRSIk_14_14_3_3" in df.columns: stoch = df["STOCHRSIk_14_14_3_3"].fillna(50) score += np.where(stoch < 20, 5, np.where(stoch > 80, -5, 0)) # CMF (-5 to +5) cmf = df["CMF_20"].fillna(0) score += np.where(cmf > 0.15, 5, np.where(cmf < -0.15, -5, 0)) # Multi-TF RSI (-5 to +5) if "RSI_14_1h" in df.columns and "RSI_14_4h" in df.columns: r1h = df["RSI_14_1h"].fillna(50) r4h = df["RSI_14_4h"].fillna(50) score += np.where((r1h < 35) & (r4h < 40), 5, np.where((r1h > 70) & (r4h > 65), -5, 0)) # BTC correlation (-5 to +3) if "btc_RSI_14_1h" in df.columns: btc_rsi = df["btc_RSI_14_1h"].fillna(50) score += np.where(btc_rsi < 30, -5, np.where(btc_rsi > 60, 3, 0)) # Volume confirmation (-3 to +3) vol_ratio = np.where(df["volume_sma_25"] > 0, df["volume"] / df["volume_sma_25"], 1.0) score += np.where((vol_ratio > 1.5) & (score > 50), 3, np.where((vol_ratio > 1.5) & (score < 50), -3, 0)) # ADX trend strength (-5 to +5) try: score += _get_adx_cdl().adx_ta_score_component(df) except Exception: pass if "nautilus_signal_score" in df.columns: score = score + pd.to_numeric(df["nautilus_signal_score"], errors="coerce").fillna(0.0) return score.clip(0, 100) # ------------------------------------------------------------------------- # Populate entry trend (vectorized — fast) # ------------------------------------------------------------------------- def populate_entry_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[:, "enter_long"] = 0 df.loc[:, "enter_tag"] = "" if "RSI_14" not in df.columns: df.loc[:, "enter_short"] = 0 return df # --- BTC trend profile: single rule, long-only on BTC (no sentiment / swing / ORB stack) --- if btc_trend_regime.sygnif_profile() == "btc_trend" and btc_trend_regime.is_btc_pair( metadata.get("pair", "") ): df.loc[:, "enter_short"] = 0 if len(df) > 0: prot = df.get("protections_long_global", pd.Series(True, index=df.index)) empty_ok = df.get("num_empty_288", pd.Series(0, index=df.index)).fillna(0) <= 60 last_prot = prot.iloc[-1] if hasattr(prot, "iloc") else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, "iloc") else True last = df.iloc[-1] if last_prot and last_empty and btc_trend_regime.btc_trend_long_row(last): df.iloc[-1, df.columns.get_loc("enter_long")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "btc_trend_long" return df # Global protections prot = df.get("protections_long_global", pd.Series(True, index=df.index)) empty_ok = df.get("num_empty_288", pd.Series(0, index=df.index)).fillna(0) <= 60 # TA score for all rows ta_score = self._calculate_ta_score_vectorized(df) # Strong TA — TA >= min + volume; skip chasing when 1h RSI is extremely overbought vol_ok = df["volume"] > (df["volume_sma_25"] * self.vol_strong_mult) r1h = df.get("RSI_14_1h", pd.Series(50.0, index=df.index)).fillna(50.0) strong_ta_htf_ok = r1h < 72.0 strong = ( prot & empty_ok & (ta_score >= self.strong_ta_min_score) & vol_ok & strong_ta_htf_ok ) df.loc[strong, "enter_long"] = 1 df.loc[strong, "enter_tag"] = "strong_ta" # --- Claude sentiment long — ambiguous zone (TA 40-64), LAST candle only --- if len(df) > 0 and not df.iloc[-1].get("enter_long", 0): last_score = ta_score.iloc[-1] last_prot = prot.iloc[-1] if hasattr(prot, 'iloc') else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, 'iloc') else True if last_prot and last_empty and self.claude_long_score_low <= last_score <= self.claude_long_score_high: pair = metadata.get("pair", "XRP/USDT") token = pair.split("/")[0] price = df.iloc[-1]["close"] headlines = self.sentiment.fetch_news(token) sentiment = self.sentiment.analyze_sentiment(token, price, last_score, headlines) # sentiment=None → API broken, skip sentiment entry (don't enter blind) # |sentiment| < sentiment_entry_min_abs → weak / not worth a slot (was 2 until 2026-04) # |sentiment| >= sentiment_entry_min_abs → combine with TA if sentiment is not None and abs(sentiment) >= self.sentiment_entry_min_abs: final_score = last_score + sentiment if final_score >= self.sentiment_threshold_buy: df.iloc[-1, df.columns.get_loc("enter_long")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = f"sygnif_s{sentiment:.0f}" # --- Failure Swing entries (last candle only) --- if len(df) > 0 and not df.iloc[-1].get("enter_long", 0): last_prot = prot.iloc[-1] if hasattr(prot, 'iloc') else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, 'iloc') else True sf_long = df.iloc[-1].get("sf_long", False) if last_prot and last_empty and sf_long: last_score = ta_score.iloc[-1] split = float(self.sf_ta_split) if last_score >= split: # sygnif_swing: failure swing + TA confluence df.iloc[-1, df.columns.get_loc("enter_long")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "sygnif_swing" else: # swing_failure: standalone, TA not confirming but pattern is clear df.iloc[-1, df.columns.get_loc("enter_long")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "swing_failure" # --- Session ORB long (BTC/ETH, last candle; uses ``orb_*`` columns when enabled) --- if len(df) > 0 and int(getattr(self, "orb_entry_enabled", 0)) and not df.iloc[-1].get("enter_long", 0): try: _mso = _get_market_sessions_orb() pair = metadata.get("pair", "") if _mso.is_orb_pair(pair) and "orb_break_long" in df.columns: last_prot = prot.iloc[-1] if hasattr(prot, "iloc") else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, "iloc") else True if last_prot and last_empty and bool(df.iloc[-1].get("orb_break_long")): df.iloc[-1, df.columns.get_loc("enter_long")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "orb_long" except Exception as e: logger.warning("[%s] ORB entry skipped: %s", metadata.get("pair"), e) # ===================================================================== # SHORT ENTRIES (futures only — guarded by can_short in config) # ===================================================================== df.loc[:, "enter_short"] = 0 prot_short = df.get("protections_short_global", pd.Series(True, index=df.index)) # Strong TA short — volume gate; skip when 4h is extremely oversold (violent bounce risk) vol_ok_short = df["volume"] > (df["volume_sma_25"] * self.vol_strong_mult) r4h = df.get("RSI_14_4h", pd.Series(50.0, index=df.index)).fillna(50.0) strong_short_htf_ok = r4h > 28.0 strong_short = ( prot_short & empty_ok & (ta_score <= self.strong_ta_short_max_score) & vol_ok_short & strong_short_htf_ok ) df.loc[strong_short, "enter_short"] = 1 df.loc[strong_short, "enter_tag"] = "strong_ta_short" # Ambiguous zone — Claude sentiment short on LAST candle only if len(df) > 0 and not df.iloc[-1].get("enter_short", 0): last_score = ta_score.iloc[-1] last_prot_s = prot_short.iloc[-1] if hasattr(prot_short, 'iloc') else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, 'iloc') else True if last_prot_s and last_empty and self.claude_short_score_low <= last_score <= self.claude_short_score_high: pair = metadata.get("pair", "XRP/USDT") token = pair.split("/")[0] price = df.iloc[-1]["close"] headlines = self.sentiment.fetch_news(token) sentiment = self.sentiment.analyze_sentiment(token, price, last_score, headlines) # sentiment=None → API broken, skip # |sentiment| < sentiment_entry_min_abs → weak (aligned with long branch) if sentiment is not None and abs(sentiment) >= self.sentiment_entry_min_abs: final_score = last_score + sentiment if final_score <= self.sentiment_threshold_sell: df.iloc[-1, df.columns.get_loc("enter_short")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = f"sygnif_short_s{sentiment:.0f}" # --- Failure Swing short entries (last candle only) --- if len(df) > 0 and not df.iloc[-1].get("enter_short", 0): last_prot_s = prot_short.iloc[-1] if hasattr(prot_short, 'iloc') else True last_empty = empty_ok.iloc[-1] if hasattr(empty_ok, 'iloc') else True sf_short = df.iloc[-1].get("sf_short", False) if last_prot_s and last_empty and sf_short: last_score = ta_score.iloc[-1] split = float(self.sf_ta_split) if last_score <= split: # sygnif_swing_short: failure swing + bearish TA confluence df.iloc[-1, df.columns.get_loc("enter_short")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "sygnif_swing_short" else: # swing_failure_short: standalone pattern df.iloc[-1, df.columns.get_loc("enter_short")] = 1 df.iloc[-1, df.columns.get_loc("enter_tag")] = "swing_failure_short" # Optional: log short/long protection snapshot for selected pairs (throttled: once per pair per candle). # Enable: SYGNIF_SHORT_DIAG=1 (optional SYGNIF_SHORT_DIAG_PAIRS="BTC/USDT:USDT,ETH/USDT:USDT") _diag = os.environ.get("SYGNIF_SHORT_DIAG", "").strip().lower() if _diag in ("1", "true", "yes") and len(df) > 0: raw = os.environ.get( "SYGNIF_SHORT_DIAG_PAIRS", "BTC/USDT:USDT,ETH/USDT:USDT", ) allowed = {p.strip() for p in raw.split(",") if p.strip()} pair = metadata.get("pair", "") if pair in allowed: last = df.iloc[-1] btc4 = float(last.get("btc_RSI_14_4h", float("nan"))) btc1h3 = float(last.get("btc_RSI_3_1h", float("nan"))) ps = bool(prot_short.iloc[-1]) if hasattr(prot_short, "iloc") else True pl = bool(prot.iloc[-1]) if hasattr(prot, "iloc") else True ts = float(ta_score.iloc[-1]) es = int(last.get("enter_short", 0) or 0) el = int(last.get("enter_long", 0) or 0) logger.info( "[SHORT_DIAG] pair=%s ta=%.0f prot_long=%s prot_short=%s " "btc_RSI_14_4h=%.1f btc_RSI_3_1h=%.1f enter_long=%d enter_short=%d", pair, ts, pl, ps, btc4, btc1h3, el, es, ) return df # ------------------------------------------------------------------------- # Populate exit trend (basic — main exits via custom_exit) # ------------------------------------------------------------------------- def populate_exit_trend(self, df: DataFrame, metadata: dict) -> DataFrame: df.loc[:, "exit_long"] = 0 df.loc[:, "exit_short"] = 0 return df # ------------------------------------------------------------------------- # NT risk-based position sizing (Lesson 2) # ------------------------------------------------------------------------- def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: Optional[float], max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: if not self._risk_manager: return proposed_stake equity = None if hasattr(self, "wallets") and self.wallets: equity = self.wallets.get_free("USDT") if not equity or equity <= 0: return proposed_stake is_futures = self.config.get("trading_mode", "") == "futures" doom_sl = self.stop_threshold_doom_futures if is_futures else self.stop_threshold_doom_spot sl_distance_pct = (doom_sl / leverage) if is_futures else doom_sl if self.dp: try: df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(df) > 0: atr_pct = df.iloc[-1].get("atr_pct", 0) if atr_pct > 0: atr_sl = (atr_pct / 100.0) * 2.0 sl_distance_pct = max(sl_distance_pct, atr_sl) except Exception: pass if sl_distance_pct <= 0: return proposed_stake if side == "long": sl_price = current_rate * (1 - sl_distance_pct) else: sl_price = current_rate * (1 + sl_distance_pct) position_size = self._risk_manager.calculate_position_size( equity=equity, entry=current_rate, stop_loss=sl_price, risk_pct=self.RISK_PCT, ) stake = (position_size * current_rate) / leverage stake = max(min_stake or 0, min(stake, max_stake)) logger.info( "NT sizing: %s sl_dist=%.2f%% equity=%.1f → stake=%.2f (proposed=%.2f)", pair, sl_distance_pct * 100, equity, stake, proposed_stake, ) return stake def _futures_volume_gate_passes(self, df: DataFrame) -> bool: """Micro-cap filter for futures entries. Raw ``volume_sma_25`` vs 50k works for high–base-volume alts; BTC/ETH use small base units per bar, so also allow ``vol_sma * close`` (≈ quote notional) ≥ $50k. """ if len(df) == 0 or "volume_sma_25" not in df.columns: return True last = df.iloc[-1] vol_avg = float(last.get("volume_sma_25", 0) or 0) close = float(last.get("close", 0) or 0) quote_vol = vol_avg * close if quote_vol >= 50_000: return True if vol_avg >= 50_000: return True return False # ------------------------------------------------------------------------- # DCA scale-in for high-conviction entries (Lesson 4) # ------------------------------------------------------------------------- def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: Optional[float], max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs) -> Optional[float]: tag = trade.enter_tag or "" if tag not in self.DCA_ELIGIBLE_TAGS: return None filled = trade.nr_of_successful_entries if filled > self.DCA_MAX_ENTRIES: return None if current_profit > self.DCA_DRAWDOWN_STEP: return None original_stake = trade.stake_amount / max(filled, 1) dca_stake = original_stake * self.DCA_SCALE_FACTOR dca_stake = max(min_stake or 0, min(dca_stake, max_stake)) logger.info( "DCA scale-in: %s tag=%s entries=%d profit=%.2f%% → +%.2f USDT", trade.pair, tag, filled, current_profit * 100, dca_stake, ) return dca_stake # ------------------------------------------------------------------------- # Doom cooldown — block re-entry after stoploss hit # ------------------------------------------------------------------------- def confirm_trade_entry(self, pair, order_type, amount, rate, time_in_force, current_time, entry_tag, side, **kwargs): now = time.time() # Standard 4h cooldown after any stoploss cooldown_since = self._doom_cooldown.get(pair, 0) if now - cooldown_since < self.doom_cooldown_secs: logger.info(f"Doom cooldown active for {pair}, skipping entry") return False # Escalated 24h lockout after 2+ losses within 24h losses = self._doom_loss_count.get(pair, []) recent = [t for t in losses if now - t < 86400] if len(recent) >= 2: logger.info(f"Consecutive loss lockout for {pair} ({len(recent)} losses in 24h)") return False # Slot caps per entry type tag = entry_tag or "" open_trades = Trade.get_trades_proxy(is_open=True) tier_abs = sentiment_tag_score_abs(tag) if tier_abs is not None and tier_abs < int(getattr(self, "sentiment_entry_min_abs", 3)): logger.info( "Sentiment tier gate: |score|=%d < %s, blocking new entry %s (%s)", tier_abs, getattr(self, "sentiment_entry_min_abs", 3), pair, tag, ) return False if tag == "strong_ta": count = sum(1 for t in open_trades if (t.enter_tag or "") == "strong_ta") if count >= self.max_slots_strong: logger.info(f"Strong TA slot cap: {count}/{self.max_slots_strong}, skipping {pair}") return False if tag == "strong_ta_short": count = sum(1 for t in open_trades if (t.enter_tag or "") == "strong_ta_short") if count >= self.max_slots_strong_short: logger.info( f"Strong TA short slot cap: {count}/{self.max_slots_strong_short}, skipping {pair}" ) return False if tag in self._swing_tags: count = sum(1 for t in open_trades if (t.enter_tag or "") in self._swing_tags) if count >= self.max_slots_swing: logger.info(f"Swing slot cap: {count}/{self.max_slots_swing}, skipping {pair} ({tag})") return False if tag == "orb_long": count = sum(1 for t in open_trades if (t.enter_tag or "") == "orb_long") if count >= int(getattr(self, "max_slots_orb", 2)): logger.info( "ORB slot cap: %d/%s, skipping %s", count, getattr(self, "max_slots_orb", 2), pair, ) return False if tag == "btc_trend_long": count = sum(1 for t in open_trades if (t.enter_tag or "") == "btc_trend_long") if count >= int(getattr(self, "max_slots_btc_trend", 2)): logger.info( "BTC trend slot cap: %d/%s, skipping %s", count, getattr(self, "max_slots_btc_trend", 2), pair, ) return False # Premium-tag slot reservation # Non-premium tags are hard-capped at `premium_nonreserved_max` open trades, # leaving the top slots available only for high-edge tags (e.g. sygnif_s-5). # Premium tags bypass this cap and may fill up to max_open_trades. if not tag_bypasses_premium_reserve(tag, self.PREMIUM_TAGS): total_open = len(open_trades) if total_open >= self.premium_nonreserved_max: logger.info( f"Premium reserve: {total_open}/{self.premium_nonreserved_max} non-premium " f"slots full, blocking {tag} on {pair} (reserved for {self.PREMIUM_TAGS})" ) return False # NT Lesson 5: global volume regime gate (premium tags bypass) if (self.config.get("trading_mode", "") == "futures" and not tag_bypasses_premium_reserve(tag, self.PREMIUM_TAGS) and tag not in getattr(self, "_tags_bypass_volume_regime", frozenset()) and self._active_volume_pairs < self.MIN_ACTIVE_VOLUME_PAIRS): logger.info( "Volume regime: only %d/%d pairs active, blocking %s on %s", self._active_volume_pairs, self.MIN_ACTIVE_VOLUME_PAIRS, tag, pair, ) return False # Futures: minimum average volume gate (filter micro caps, swing bypasses) if (self.config.get("trading_mode", "") == "futures" and tag not in self._swing_tags and tag not in getattr(self, "_tags_bypass_volume_regime", frozenset()) and self.dp): df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(df) > 0 and not self._futures_volume_gate_passes(df): last = df.iloc[-1] vol_avg = float(last.get("volume_sma_25", 0) or 0) close = float(last.get("close", 0) or 0) logger.info( "Futures volume gate: %s vol_sma_25=%.0f quote~%.0f USDT below threshold, skipping", pair, vol_avg, vol_avg * close, ) return False return True def confirm_trade_exit(self, pair, trade, order_type, amount, rate, time_in_force, exit_reason, current_time, **kwargs): reason = exit_reason.lower() is_loss_exit = ( "stoploss" in reason # stoploss_on_exchange, exit_stoploss_conditional or "stop_loss" in reason # trailing_stop_loss or "vol_sl" in reason # exit_sf_vol_sl, exit_sf_short_vol_sl ) if is_loss_exit: now = time.time() self._doom_cooldown[pair] = now self._save_doom_cooldown() losses = self._doom_loss_count.get(pair, []) losses = [t for t in losses if now - t < 86400] + [now] self._doom_loss_count[pair] = losses logger.info(f"Doom cooldown set for {pair} after {exit_reason} ({len(losses)} losses in 24h)") # NT Lesson 3: emit position_closed with the last active SL tier if hasattr(self, "_event_log") and self._event_log is not None: try: trade_id = getattr(trade, "id", None) last_tier = self._last_sl_tier.pop(trade_id, "unknown") if trade_id else "unknown" leverage = trade.leverage or 1.0 realized_pnl = (rate - trade.open_rate) * amount if trade.is_short: realized_pnl = -realized_pnl self._event_log.emit_position_closed( instrument_id=pair, entry_side="short" if trade.is_short else "long", avg_px_open=trade.open_rate, avg_px_close=rate, realized_pnl=realized_pnl, trade_id=trade_id, exit_reason=exit_reason, last_sl_tier=last_tier, leverage=leverage, enter_tag=trade.enter_tag or "", ) except Exception: pass return True # ------------------------------------------------------------------------- # Custom exit — NFI-style profit-tiered exits # ------------------------------------------------------------------------- def custom_exit( self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs ): df, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(df) < 2: return None last = df.iloc[-1] prev = df.iloc[-2] filled_entries = trade.select_filled_orders(trade.entry_side) if not filled_entries: return None is_short = trade.is_short leverage = trade.leverage or 1.0 # ================================================================== # SHORT EXITS # ================================================================== if is_short: return self._custom_exit_short( last, prev, current_profit, current_rate, trade, filled_entries, leverage, ) # ================================================================== # BTC risk-off — flatten small / flat longs when BTC 1h confirms a spill # (uses merged BTC columns on alts; skipped when columns absent e.g. BTC pair) # ================================================================== try: btc_ch = float(last.get("btc_change_pct_1h", float("nan"))) btc_r1 = float(last.get("btc_RSI_14_1h", 50.0)) btc_r4 = float(last.get("btc_RSI_14_4h", 50.0)) except (TypeError, ValueError): btc_ch = float("nan") btc_r1 = 50.0 btc_r4 = 50.0 if not pd.isna(btc_ch) and "btc_change_pct_1h" in last.index: profit_cap = 0.035 if leverage <= 1.0 else min(0.14, 0.035 * leverage) if ( btc_ch <= -1.15 and btc_r1 < 37.0 and btc_r4 < 46.0 and current_profit < profit_cap ): return "exit_btc_risk_off" tpx = self._sygnif_exit_trailing_take_profit(trade, current_rate) if tpx: return tpx # ================================================================== # FAILURE SWING EXITS (tag-based routing) # ================================================================== enter_tag = trade.enter_tag or "" if enter_tag == "swing_failure": return self._exit_swing_failure(last, current_rate, trade, current_profit) if enter_tag in self._swing_hybrid_long: # Hybrid: check both EMA-TP and Williams %R, first one wins sf_exit = self._exit_swing_failure(last, current_rate, trade, current_profit) if sf_exit: return sf_exit # Fall through to Williams %R below # ================================================================== # LONG EXITS # ================================================================== rsi14 = last.get("RSI_14", 50) rsi14_1h = last.get("RSI_14_1h", 50) # Minimum profit before RSI exits activate (don't exit tiny gains) min_profit_for_rsi = 0.02 if leverage <= 1.0 else 0.02 * leverage if current_profit >= min_profit_for_rsi: above_ema200 = last["close"] > last.get("EMA_200", 0) rsi_threshold = self._get_exit_rsi_threshold(current_profit, above_ema200, leverage) if rsi14 < rsi_threshold: return f"exit_profit_rsi_{current_profit:.1%}" # --- Williams %R reversal exit: was at peak, now falling --- # Leverage-normalized profit gate matches RSI exit pattern min_profit_for_willr = 0.02 if leverage <= 1.0 else 0.02 * leverage willr = last.get("WILLR_14", -50) willr_prev = prev.get("WILLR_14", -50) if willr is not None and willr_prev is not None: willr_topped = willr_prev > -10 and willr < willr_prev if willr_topped and current_profit > min_profit_for_willr: return "exit_willr_reversal" # --- Doom stoploss now handled by custom_stoploss + stoploss_on_exchange --- # --- Soft stoploss — fires before exchange SL, needs fewer conditions --- is_futures = self.config.get("trading_mode", "") == "futures" soft_sl = -(self.stop_threshold_doom_futures * self.soft_sl_ratio_futures) if is_futures else -(self.stop_threshold_doom_spot * self.soft_sl_ratio_spot) if current_profit < soft_sl: rsi_falling = rsi14 < last.get("RSI_14_shift3", rsi14) if rsi_falling and (last["close"] < last.get("EMA_200", float("inf")) or rsi14 > prev.get("RSI_14", 50)): return "exit_stoploss_conditional" return None # ------------------------------------------------------------------------- # Failure Swing exit — Heavy91 EMA-TP + volatility-adjusted SL # ------------------------------------------------------------------------- def _exit_swing_failure(self, last, current_rate, trade, current_profit): ema_tp = last.get("sf_tp_ema", 0) sl_pct = last.get("sf_sl_pct", 0.02) is_short = trade.is_short if not is_short: # Long TP: price reaches volatility-adjusted EMA if ema_tp and current_rate >= ema_tp and current_profit > 0.005: return "exit_sf_ema_tp" # Long SL: dynamic volatility-adjusted stop entry = trade.open_rate sl_price = entry * (1 - sl_pct) if current_rate <= sl_price: return "exit_sf_vol_sl" else: # Short TP: price falls to volatility-adjusted EMA if ema_tp and current_rate <= ema_tp and current_profit > 0.005: return "exit_sf_short_ema_tp" # Short SL: dynamic volatility-adjusted stop entry = trade.open_rate sl_price = entry * (1 + sl_pct) if current_rate >= sl_price: return "exit_sf_short_vol_sl" return None # ------------------------------------------------------------------------- # Short exit logic — NFI-style inverted signals # ------------------------------------------------------------------------- def _custom_exit_short(self, last, prev, current_profit, current_rate, trade, filled_entries, leverage): tpx = self._sygnif_exit_trailing_take_profit(trade, current_rate) if tpx: return tpx enter_tag = trade.enter_tag or "" # --- Failure Swing short exits --- if enter_tag == "swing_failure_short": return self._exit_swing_failure(last, current_rate, trade, current_profit) if enter_tag in self._swing_hybrid_short: sf_exit = self._exit_swing_failure(last, current_rate, trade, current_profit) if sf_exit: return sf_exit # Fall through to Williams %R below rsi14 = last.get("RSI_14", 50) rsi14_1h = last.get("RSI_14_1h", 50) # --- Profit-tiered RSI exit for shorts (inverted) --- min_profit_for_rsi = 0.02 if leverage <= 1.0 else 0.02 * leverage if current_profit >= min_profit_for_rsi: below_ema200 = last["close"] < last.get("EMA_200", float("inf")) rsi_threshold = self._get_short_exit_rsi_threshold(current_profit, below_ema200, leverage) if rsi14 > rsi_threshold: return f"exit_short_profit_rsi_{current_profit:.1%}" # --- Williams %R reversal exit for shorts: was at bottom, now rising --- # Leverage-normalized profit gate matches RSI exit pattern min_profit_for_willr = 0.02 if leverage <= 1.0 else 0.02 * leverage willr = last.get("WILLR_14", -50) willr_prev = prev.get("WILLR_14", -50) if willr is not None and willr_prev is not None: willr_bottomed = willr_prev < -90 and willr > willr_prev if willr_bottomed and current_profit > min_profit_for_willr: return "exit_short_willr_reversal" # --- Doom stoploss now handled by custom_stoploss + stoploss_on_exchange --- # --- Soft stoploss for shorts — fires before exchange SL --- is_futures = self.config.get("trading_mode", "") == "futures" soft_sl = -(self.stop_threshold_doom_futures * self.soft_sl_ratio_futures) if is_futures else -(self.stop_threshold_doom_spot * self.soft_sl_ratio_spot) if current_profit < soft_sl: rsi_rising = rsi14 > last.get("RSI_14_shift3", rsi14) if rsi_rising and (last["close"] > last.get("EMA_200", 0) or rsi14 < prev.get("RSI_14", 50)): return "exit_short_stoploss_conditional" return None # ------------------------------------------------------------------------- # Profit-tiered RSI threshold (from NFI long_exit_main) # ------------------------------------------------------------------------- def _get_exit_rsi_threshold(self, profit: float, above_ema200: bool, leverage: float = 1.0) -> float: """ Higher profit → more willing to exit on RSI dip. Below EMA200 → stricter (lower RSI required to exit). Leverage-aware: at 3x, 1% profit = 0.33% price move — too thin to exit on. """ # Normalize profit to price-move equivalent for leveraged trades adj_profit = profit / leverage if leverage > 1.0 else profit offset = 0 if above_ema200 else 2 if adj_profit < 0.01: return 10.0 + offset elif adj_profit < 0.02: return 28.0 + offset elif adj_profit < 0.03: return 30.0 + offset elif adj_profit < 0.04: return 32.0 + offset elif adj_profit < 0.05: return 34.0 + offset elif adj_profit < 0.06: return 36.0 + offset elif adj_profit < 0.08: return 38.0 + offset elif adj_profit < 0.10: return 42.0 + offset elif adj_profit < 0.12: return 46.0 + offset elif adj_profit < 0.20: return 44.0 + offset else: return 42.0 + offset # ------------------------------------------------------------------------- # Short exit RSI threshold — inverted (high RSI = cover short) # NFI short_exit_main pattern # ------------------------------------------------------------------------- def _get_short_exit_rsi_threshold(self, profit: float, below_ema200: bool, leverage: float = 1.0) -> float: adj_profit = profit / leverage if leverage > 1.0 else profit offset = 0 if below_ema200 else -2 if adj_profit < 0.01: return 90.0 + offset elif adj_profit < 0.02: return 72.0 + offset elif adj_profit < 0.03: return 70.0 + offset elif adj_profit < 0.04: return 68.0 + offset elif adj_profit < 0.05: return 66.0 + offset elif adj_profit < 0.06: return 64.0 + offset elif adj_profit < 0.08: return 62.0 + offset elif adj_profit < 0.10: return 58.0 + offset elif adj_profit < 0.12: return 54.0 + offset elif adj_profit < 0.20: return 56.0 + offset else: return 58.0 + offset # --------------------------------------------------------------------------- # Freqtrade entrypoint: SygnifStrategy name → default stack or MarketStrategy2 # --------------------------------------------------------------------------- def _import_market_strategy2_class(): """Resolve MarketStrategy2 whether this file lives in ``user_data/strategies`` or repo root.""" import sys here = Path(__file__).resolve().parent paths = [here] nested = here / "user_data" / "strategies" if (nested / "MarketStrategy2.py").is_file(): paths.append(nested) for p in paths: s = str(p) if s not in sys.path: sys.path.insert(0, s) try: from MarketStrategy2 import MarketStrategy2 as ms2 # noqa: E402 return ms2 except ImportError as err: logger.error("MarketStrategy2 import failed (%s)", err) return None _MS2_BASE = None if _sygnif_strategy_backend() == "ms2": _MS2_BASE = _import_market_strategy2_class() if _MS2_BASE is None: logger.error( "SYGNIF_STRATEGY_BACKEND=ms2 but MarketStrategy2 is unavailable; " "using default Sygnif stack.", ) if _MS2_BASE is not None: logger.info( "SygnifStrategy: using MarketStrategy2 backend (SYGNIF_STRATEGY_BACKEND=%r)", os.environ.get("SYGNIF_STRATEGY_BACKEND", ""), ) class SygnifStrategy(_MS2_BASE): """ Same as :class:`MarketStrategy2` while keeping config ``strategy: SygnifStrategy``. Enable with ``SYGNIF_STRATEGY_BACKEND=ms2`` or ``marketstrategy2``. Restart Freqtrade after changing — import-time switch. """ pass else: class SygnifStrategy(Github_Giansn_SYGNIF__SygnifStrategy__20260428_213520): """Default: local HTTP / legacy cloud sentiment (see module docstring).""" pass