from __future__ import annotations import json import os from dataclasses import dataclass, field from pathlib import Path from typing import Any PROJECT_ROOT = Path(__file__).resolve().parent DEFAULT_RUNTIME_DIR = Path.home() / ".local" / "state" / "CryptoTradingBot" def _env_bool(name: str, default: bool) -> bool: value = os.getenv(name) if value is None: return default return value.strip().lower() in {"1", "true", "yes", "on", "y"} def _coerce_path(value: str | Path | None, default: Path) -> Path: if value is None or str(value).strip() == "": return default return Path(value).expanduser().resolve() def _env_number(name: str, default: int | float) -> int | float: value = os.getenv(name) if value is None or value.strip() == "": return default if isinstance(default, int) and not isinstance(default, bool): return int(value) return float(value) def _env_tuple(name: str, default: tuple[str, ...]) -> tuple[str, ...]: value = os.getenv(name) if value is None: return default return tuple(part.strip().upper() for part in value.split(",") if part.strip()) @dataclass(frozen=True) class BotConfig: max_total_trades: int = 3 default_leverage: int = 5 leverage_by_coin: dict[str, float] = field(default_factory=dict) crash_roe_trigger_pct: float = 15.0 trade_size_usd: float = 50.0 wallet_equity_usdc: float = 500.0 risk_per_trade_pct: float = 0.75 position_sizing_mode: str = "fixed" max_position_margin_pct: float = 20.0 max_position_notional_pct: float = 60.0 min_order_notional_usd: float = 12.0 crash_window_minutes: int = 15 flash_crash_trigger_pct: float = 4.5 atr_sl_multiplier: float = 1.5 break_even_activation_pct: float = 0.4 v_shape_activation_pct: float = 1.5 v_shape_trail_dist_pct: float = 0.75 dead_fish_time_limit_mins: int = 15 max_hard_stop_pct: float = 3.0 min_volume_24h: float = 10_000_000.0 cooldown_minutes: int = 60 max_daily_trades: int = 10 max_daily_loss_usd: float = 25.0 blacklist: tuple[str, ...] = ("FTT", "LUNA", "USTC", "STABLE") allowed_coins: tuple[str, ...] = () max_scan_coins: int = 60 baseline_cache_ttl_seconds: int = 300 error_log_throttle_seconds: int = 300 dry_run: bool = True paper_trading: bool = False runtime_dir: Path = field(default_factory=lambda: DEFAULT_RUNTIME_DIR) tls_verify: bool = True strategy_family: str = "flash_crash" squeeze_lookback_ticks: int = 24 squeeze_max_band_width_pct: float = 2.5 squeeze_breakout_pct: float = 1.2 trend_pullback_min_rsi: float = 35.0 trend_pullback_max_rsi: float = 70.0 trend_pullback_max_distance_pct: float = 1.5 mean_reversion_max_rsi: float = 35.0 mean_reversion_band_tolerance_pct: float = 0.3 multi_day_min_trend_pct: float = 2.0 multi_day_max_pullback_distance_pct: float = 3.0 relative_strength_lookback_ticks: int = 24 relative_strength_top_n: int = 4 relative_strength_min_momentum_pct: float = 3.0 relative_strength_min_positive_candidates: int = 3 @classmethod def from_file(cls, path: str | Path = PROJECT_ROOT / "strategy_config.json") -> "BotConfig": raw: dict[str, Any] = {} config_path = Path(path) if config_path.exists(): raw = json.loads(config_path.read_text(encoding="utf-8")) params = raw.get("parameters", {}) if isinstance(raw.get("parameters", {}), dict) else {} filters = raw.get("filters", {}) if isinstance(raw.get("filters", {}), dict) else {} blacklist = raw.get("blacklist", cls.blacklist) whitelist = raw.get("whitelist", ()) cfg = cls( max_total_trades=int(params.get("max_total_trades", cls.max_total_trades)), default_leverage=int(params.get("default_leverage", cls.default_leverage)), leverage_by_coin={str(k).upper(): float(v) for k, v in params.get("leverage_by_coin", {}).items()}, crash_roe_trigger_pct=float(params.get("crash_roe_trigger_pct", cls.crash_roe_trigger_pct)), trade_size_usd=float(params.get("trade_size_usd", cls.trade_size_usd)), wallet_equity_usdc=float(params.get("wallet_equity_usdc", cls.wallet_equity_usdc)), risk_per_trade_pct=float(params.get("risk_per_trade_pct", cls.risk_per_trade_pct)), position_sizing_mode=str(params.get("position_sizing_mode", cls.position_sizing_mode)), max_position_margin_pct=float(params.get("max_position_margin_pct", cls.max_position_margin_pct)), max_position_notional_pct=float(params.get("max_position_notional_pct", cls.max_position_notional_pct)), min_order_notional_usd=float(params.get("min_order_notional_usd", cls.min_order_notional_usd)), crash_window_minutes=int(params.get("crash_window_minutes", cls.crash_window_minutes)), flash_crash_trigger_pct=float(params.get("flash_crash_trigger_pct", cls.flash_crash_trigger_pct)), atr_sl_multiplier=float(params.get("atr_sl_multiplier", cls.atr_sl_multiplier)), break_even_activation_pct=float(params.get("break_even_activation_pct", cls.break_even_activation_pct)), v_shape_activation_pct=float(params.get("v_shape_activation_pct", cls.v_shape_activation_pct)), v_shape_trail_dist_pct=float(params.get("v_shape_trail_dist_pct", cls.v_shape_trail_dist_pct)), dead_fish_time_limit_mins=int(params.get("dead_fish_time_limit_mins", cls.dead_fish_time_limit_mins)), max_hard_stop_pct=float(params.get("max_hard_stop_pct", cls.max_hard_stop_pct)), min_volume_24h=float(filters.get("min_volume_24h", params.get("min_volume_24h", cls.min_volume_24h))), cooldown_minutes=int(params.get("cooldown_minutes", cls.cooldown_minutes)), max_daily_trades=int(params.get("max_daily_trades", cls.max_daily_trades)), max_daily_loss_usd=float(params.get("max_daily_loss_usd", cls.max_daily_loss_usd)), blacklist=tuple(str(x).upper() for x in blacklist), allowed_coins=tuple(str(x).upper() for x in params.get("allowed_coins", filters.get("allowed_coins", whitelist))), max_scan_coins=int(params.get("max_scan_coins", filters.get("max_scan_coins", cls.max_scan_coins))), baseline_cache_ttl_seconds=int(params.get("baseline_cache_ttl_seconds", cls.baseline_cache_ttl_seconds)), error_log_throttle_seconds=int(params.get("error_log_throttle_seconds", cls.error_log_throttle_seconds)), dry_run=bool(params.get("dry_run", cls.dry_run)), paper_trading=bool(params.get("paper_trading", cls.paper_trading)), runtime_dir=_coerce_path(params.get("runtime_dir"), DEFAULT_RUNTIME_DIR), tls_verify=bool(params.get("tls_verify", cls.tls_verify)), strategy_family=str(params.get("strategy_family", cls.strategy_family)), squeeze_lookback_ticks=int(params.get("squeeze_lookback_ticks", cls.squeeze_lookback_ticks)), squeeze_max_band_width_pct=float(params.get("squeeze_max_band_width_pct", cls.squeeze_max_band_width_pct)), squeeze_breakout_pct=float(params.get("squeeze_breakout_pct", cls.squeeze_breakout_pct)), trend_pullback_min_rsi=float(params.get("trend_pullback_min_rsi", cls.trend_pullback_min_rsi)), trend_pullback_max_rsi=float(params.get("trend_pullback_max_rsi", cls.trend_pullback_max_rsi)), trend_pullback_max_distance_pct=float(params.get("trend_pullback_max_distance_pct", cls.trend_pullback_max_distance_pct)), mean_reversion_max_rsi=float(params.get("mean_reversion_max_rsi", cls.mean_reversion_max_rsi)), mean_reversion_band_tolerance_pct=float(params.get("mean_reversion_band_tolerance_pct", cls.mean_reversion_band_tolerance_pct)), multi_day_min_trend_pct=float(params.get("multi_day_min_trend_pct", cls.multi_day_min_trend_pct)), multi_day_max_pullback_distance_pct=float(params.get("multi_day_max_pullback_distance_pct", cls.multi_day_max_pullback_distance_pct)), relative_strength_lookback_ticks=int(params.get("relative_strength_lookback_ticks", cls.relative_strength_lookback_ticks)), relative_strength_top_n=int(params.get("relative_strength_top_n", cls.relative_strength_top_n)), relative_strength_min_momentum_pct=float(params.get("relative_strength_min_momentum_pct", cls.relative_strength_min_momentum_pct)), relative_strength_min_positive_candidates=int(params.get("relative_strength_min_positive_candidates", cls.relative_strength_min_positive_candidates)), ) return cls( **{ **cfg.__dict__, "max_total_trades": _env_number("CTB_MAX_TOTAL_TRADES", cfg.max_total_trades), "default_leverage": _env_number("CTB_DEFAULT_LEVERAGE", cfg.default_leverage), "leverage_by_coin": cfg.leverage_by_coin, "crash_roe_trigger_pct": _env_number("CTB_CRASH_ROE_TRIGGER_PCT", cfg.crash_roe_trigger_pct), "trade_size_usd": _env_number("CTB_TRADE_SIZE_USD", cfg.trade_size_usd), "wallet_equity_usdc": _env_number("CTB_WALLET_EQUITY_USDC", cfg.wallet_equity_usdc), "risk_per_trade_pct": _env_number("CTB_RISK_PER_TRADE_PCT", cfg.risk_per_trade_pct), "position_sizing_mode": os.getenv("CTB_POSITION_SIZING_MODE", cfg.position_sizing_mode), "max_position_margin_pct": _env_number("CTB_MAX_POSITION_MARGIN_PCT", cfg.max_position_margin_pct), "max_position_notional_pct": _env_number("CTB_MAX_POSITION_NOTIONAL_PCT", cfg.max_position_notional_pct), "min_order_notional_usd": _env_number("CTB_MIN_ORDER_NOTIONAL_USD", cfg.min_order_notional_usd), "crash_window_minutes": _env_number("CTB_CRASH_WINDOW_MINUTES", cfg.crash_window_minutes), "flash_crash_trigger_pct": _env_number("CTB_FLASH_CRASH_TRIGGER_PCT", cfg.flash_crash_trigger_pct), "atr_sl_multiplier": _env_number("CTB_ATR_SL_MULTIPLIER", cfg.atr_sl_multiplier), "break_even_activation_pct": _env_number("CTB_BREAK_EVEN_ACTIVATION_PCT", cfg.break_even_activation_pct), "v_shape_activation_pct": _env_number("CTB_V_SHAPE_ACTIVATION_PCT", cfg.v_shape_activation_pct), "v_shape_trail_dist_pct": _env_number("CTB_V_SHAPE_TRAIL_DIST_PCT", cfg.v_shape_trail_dist_pct), "dead_fish_time_limit_mins": _env_number("CTB_DEAD_FISH_TIME_LIMIT_MINS", cfg.dead_fish_time_limit_mins), "max_hard_stop_pct": _env_number("CTB_MAX_HARD_STOP_PCT", cfg.max_hard_stop_pct), "min_volume_24h": _env_number("CTB_MIN_VOLUME_24H", cfg.min_volume_24h), "cooldown_minutes": _env_number("CTB_COOLDOWN_MINUTES", cfg.cooldown_minutes), "max_daily_trades": _env_number("CTB_MAX_DAILY_TRADES", cfg.max_daily_trades), "max_daily_loss_usd": _env_number("CTB_MAX_DAILY_LOSS_USD", cfg.max_daily_loss_usd), "allowed_coins": _env_tuple("CTB_ALLOWED_COINS", cfg.allowed_coins), "max_scan_coins": _env_number("CTB_MAX_SCAN_COINS", cfg.max_scan_coins), "baseline_cache_ttl_seconds": _env_number("CTB_BASELINE_CACHE_TTL_SECONDS", cfg.baseline_cache_ttl_seconds), "error_log_throttle_seconds": _env_number("CTB_ERROR_LOG_THROTTLE_SECONDS", cfg.error_log_throttle_seconds), "dry_run": _env_bool("CTB_DRY_RUN", cfg.dry_run), "paper_trading": _env_bool("CTB_PAPER_TRADING", cfg.paper_trading), "tls_verify": _env_bool("CTB_TLS_VERIFY", cfg.tls_verify), "runtime_dir": _coerce_path(os.getenv("CTB_RUNTIME_DIR"), cfg.runtime_dir), "strategy_family": os.getenv("CTB_STRATEGY_FAMILY", cfg.strategy_family), "squeeze_lookback_ticks": _env_number("CTB_SQUEEZE_LOOKBACK_TICKS", cfg.squeeze_lookback_ticks), "squeeze_max_band_width_pct": _env_number("CTB_SQUEEZE_MAX_BAND_WIDTH_PCT", cfg.squeeze_max_band_width_pct), "squeeze_breakout_pct": _env_number("CTB_SQUEEZE_BREAKOUT_PCT", cfg.squeeze_breakout_pct), "trend_pullback_min_rsi": _env_number("CTB_TREND_PULLBACK_MIN_RSI", cfg.trend_pullback_min_rsi), "trend_pullback_max_rsi": _env_number("CTB_TREND_PULLBACK_MAX_RSI", cfg.trend_pullback_max_rsi), "trend_pullback_max_distance_pct": _env_number("CTB_TREND_PULLBACK_MAX_DISTANCE_PCT", cfg.trend_pullback_max_distance_pct), "mean_reversion_max_rsi": _env_number("CTB_MEAN_REVERSION_MAX_RSI", cfg.mean_reversion_max_rsi), "mean_reversion_band_tolerance_pct": _env_number("CTB_MEAN_REVERSION_BAND_TOLERANCE_PCT", cfg.mean_reversion_band_tolerance_pct), "multi_day_min_trend_pct": _env_number("CTB_MULTI_DAY_MIN_TREND_PCT", cfg.multi_day_min_trend_pct), "multi_day_max_pullback_distance_pct": _env_number("CTB_MULTI_DAY_MAX_PULLBACK_DISTANCE_PCT", cfg.multi_day_max_pullback_distance_pct), "relative_strength_lookback_ticks": _env_number("CTB_RELATIVE_STRENGTH_LOOKBACK_TICKS", cfg.relative_strength_lookback_ticks), "relative_strength_top_n": _env_number("CTB_RELATIVE_STRENGTH_TOP_N", cfg.relative_strength_top_n), "relative_strength_min_momentum_pct": _env_number("CTB_RELATIVE_STRENGTH_MIN_MOMENTUM_PCT", cfg.relative_strength_min_momentum_pct), "relative_strength_min_positive_candidates": _env_number("CTB_RELATIVE_STRENGTH_MIN_POSITIVE_CANDIDATES", cfg.relative_strength_min_positive_candidates), } ) @dataclass(frozen=True) class RuntimePaths: runtime_dir: Path tradeanalyse_dir: Path reports_dir: Path trading_log: Path trade_journal: Path cooldowns: Path portfolio_state: Path account_health: Path profit_history: Path crash_report: Path kill_switch: Path paper_state: Path market_context: Path near_miss_journal: Path @classmethod def from_config(cls, cfg: BotConfig) -> "RuntimePaths": runtime_dir = cfg.runtime_dir tradeanalyse_dir = runtime_dir / "Tradeanalyse" return cls( runtime_dir=runtime_dir, tradeanalyse_dir=tradeanalyse_dir, reports_dir=runtime_dir / "reports", trading_log=tradeanalyse_dir / "trading_log.txt", trade_journal=tradeanalyse_dir / "trade_journal.jsonl", cooldowns=tradeanalyse_dir / "cooldowns_v61.json", portfolio_state=runtime_dir / "portfolio_state.json", account_health=runtime_dir / "account_health.txt", profit_history=runtime_dir / "profit_history.log", crash_report=runtime_dir / "crash_report.log", kill_switch=runtime_dir / "KILL_SWITCH", paper_state=runtime_dir / "paper_state.json", market_context=runtime_dir / "market_context.jsonl", near_miss_journal=tradeanalyse_dir / "near_miss.jsonl", ) def ensure_dirs(self) -> None: self.runtime_dir.mkdir(parents=True, exist_ok=True) self.tradeanalyse_dir.mkdir(parents=True, exist_ok=True) self.reports_dir.mkdir(parents=True, exist_ok=True)