"""Multi-timeframe combo presets for Crypto Wyckoff Screener. Roles (engine rule aliases stay D/W/M): high → Cycle (rules as 1M) mid → Phase (rules as 1w) low → Event (rules as 1d) Actual bar TFs come from the combo (e.g. 8h/4h/1h). """ from __future__ import annotations import json import re import threading from copy import deepcopy from pathlib import Path from typing import Any from crypto_wyckoff.io import DATA_DIR, ensure_dirs ROLE_LOW = "1d" ROLE_MID = "1w" ROLE_HIGH = "1M" # Minutes for ordering / validation (provider labels) _TF_MINUTES: dict[str, int] = { "1m": 1, "2m": 2, "3m": 3, "4m": 4, "5m": 5, "10m": 10, "15m": 15, "20m": 20, "25m": 25, "30m": 30, "45m": 45, "1h": 60, "2h": 120, "3h": 180, "4h": 240, "5h": 300, "6h": 360, "7h": 420, "8h": 480, "9h": 540, "10h": 600, "11h": 660, "12h": 720, "16h": 960, "20h": 1200, "1d": 1440, "2d": 2880, "3d": 4320, "4d": 5760, "5d": 7200, "6d": 8640, "1w": 10080, "2w": 20160, "3w": 30240, "1M": 43200, } # TFs we allow in custom combos (provider-backed + local 1M) ALLOWED_TFS: tuple[str, ...] = ( "1h", "2h", "3h", "4h", "6h", "8h", "12h", "1d", "2d", "3d", "1w", "1M", ) BUILTIN: list[dict[str, Any]] = [ { "id": "h8_4_1", "label": "8h / 4h / 1h", "high": "8h", "mid": "4h", "low": "1h", "builtin": True, }, { "id": "d_w_m", "label": "1d / 1w / 1M", "high": "1M", "mid": "1w", "low": "1d", "builtin": True, }, ] _COMBOS_FILE = DATA_DIR / "combos.json" _lock = threading.Lock() _cache: list[dict[str, Any]] | None = None def tf_minutes(tf: str) -> int | None: if tf in _TF_MINUTES: return _TF_MINUTES[tf] # tolerate provider typo "10" → skip m = re.fullmatch(r"(\d+)([mhdwM])", tf) if not m: return None n, u = int(m.group(1)), m.group(2) mult = {"m": 1, "h": 60, "d": 1440, "w": 10080, "M": 43200}[u] return n * mult def combo_id_for(high: str, mid: str, low: str) -> str: def _tok(t: str) -> str: return t.replace("/", "_") return f"{_tok(high)}_{_tok(mid)}_{_tok(low)}" def validate_combo(high: str, mid: str, low: str) -> str | None: """Return error message or None if ok.""" for tf in (high, mid, low): if tf not in ALLOWED_TFS: return f"不支持的周期: {tf}" if len({high, mid, low}) < 3: return "高/中/低周期必须互不相同" hm, mm, lm = tf_minutes(high), tf_minutes(mid), tf_minutes(low) if hm is None or mm is None or lm is None: return "无法解析周期长度" if not (hm > mm > lm): return "须满足 高 > 中 > 低（例如 8h > 4h > 1h）" return None def _normalize(row: dict[str, Any]) -> dict[str, Any] | None: high, mid, low = row.get("high"), row.get("mid"), row.get("low") if not high or not mid or not low: return None err = validate_combo(str(high), str(mid), str(low)) if err: return None cid = str(row.get("id") or combo_id_for(high, mid, low)) label = str(row.get("label") or f"{high} / {mid} / {low}") return { "id": cid, "label": label, "high": str(high), "mid": str(mid), "low": str(low), "builtin": bool(row.get("builtin", False)), } def _load_raw() -> list[dict[str, Any]]: ensure_dirs() if not _COMBOS_FILE.exists(): return deepcopy(BUILTIN) try: data = json.loads(_COMBOS_FILE.read_text(encoding="utf-8")) items = data.get("combos") if isinstance(data, dict) else data if not isinstance(items, list): return deepcopy(BUILTIN) except (OSError, json.JSONDecodeError): return deepcopy(BUILTIN) out: list[dict[str, Any]] = [] seen: set[str] = set() for b in BUILTIN: out.append(deepcopy(b)) seen.add(b["id"]) for row in items: if not isinstance(row, dict): continue norm = _normalize(row) if not norm or norm["id"] in seen: continue if norm["id"] in {b["id"] for b in BUILTIN}: continue norm["builtin"] = False out.append(norm) seen.add(norm["id"]) return out def _save(combos: list[dict[str, Any]]) -> None: ensure_dirs() custom = [c for c in combos if not c.get("builtin")] payload = {"combos": custom} tmp = _COMBOS_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(_COMBOS_FILE) def list_combos() -> list[dict[str, Any]]: global _cache with _lock: if _cache is None: _cache = _load_raw() return deepcopy(_cache) def get_combo(combo_id: str | None) -> dict[str, Any]: combos = list_combos() if combo_id: for c in combos: if c["id"] == combo_id: return deepcopy(c) return deepcopy(combos[0]) def add_combo(high: str, mid: str, low: str, label: str | None = None) -> dict[str, Any]: err = validate_combo(high, mid, low) if err: raise ValueError(err) cid = combo_id_for(high, mid, low) row = { "id": cid, "label": label or f"{high} / {mid} / {low}", "high": high, "mid": mid, "low": low, "builtin": False, } with _lock: combos = _load_raw() for c in combos: if c["id"] == cid or (c["high"], c["mid"], c["low"]) == (high, mid, low): _cache = combos return deepcopy(c) combos.append(row) _save(combos) _cache = combos return deepcopy(row) def delete_combo(combo_id: str) -> bool: with _lock: combos = _load_raw() kept: list[dict[str, Any]] = [] removed = False for c in combos: if c["id"] == combo_id: if c.get("builtin"): raise ValueError("内置组合不可删除") removed = True continue kept.append(c) if removed: _save(kept) _cache = kept return removed def all_tfs_for_combos(combos: list[dict[str, Any]] | None = None) -> list[str]: """Unique TFs needed by active combos (stable order).""" rows = combos if combos is not None else list_combos() seen: list[str] = [] for c in rows: for k in ("low", "mid", "high"): tf = c[k] if tf not in seen: seen.append(tf) return seen def lookback_for(tf: str) -> int: defaults = { "1h": 500, "2h": 400, "3h": 350, "4h": 300, "6h": 280, "8h": 250, "12h": 220, "1d": 250, "2d": 200, "3d": 180, "1w": 104, "1M": 60, } return defaults.get(tf, 200)