jackyu66gitandCursor
18a7f485e6
自动刷新常态只拉 recent 尾部 K，每 1 分钟全量重算缠论；修复结构区缓存导入；默认指标/4h·1h·15m/近30天；同步 ECR-009 screener 相关改动。 Co-authored-by: Cursor <cursoragent@cursor.com>
128 lines
3.7 KiB
Python
128 lines
3.7 KiB
Python
"""Background tip + scan scheduler for crypto wyckoff (all enabled combos)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
|
|
from crypto_wyckoff.combos import all_tfs_for_combos, list_combos
|
|
from crypto_wyckoff.io import (
|
|
backfill_symbol,
|
|
bar_count,
|
|
fetch_symbols_from_provider,
|
|
tip_update_symbol,
|
|
)
|
|
from crypto_wyckoff.pipeline import analyze_and_store
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_thread: threading.Thread | None = None
|
|
_stop = threading.Event()
|
|
_status: dict = {
|
|
"running": False,
|
|
"last_tick_at": None,
|
|
"last_error": None,
|
|
"symbols_total": 0,
|
|
"symbols_scanned": 0,
|
|
"tick_interval_sec": 60,
|
|
"backfill_done": False,
|
|
}
|
|
_status_lock = threading.Lock()
|
|
|
|
|
|
def _set(**kwargs):
|
|
with _status_lock:
|
|
_status.update(kwargs)
|
|
|
|
|
|
def get_status() -> dict:
|
|
with _status_lock:
|
|
return dict(_status)
|
|
|
|
|
|
def run_tick(max_symbols: int | None = None, force_rescan: bool = False) -> dict:
|
|
"""One cycle: refresh symbols, tip-update, analyze each combo."""
|
|
symbols = fetch_symbols_from_provider()
|
|
if max_symbols:
|
|
symbols = symbols[:max_symbols]
|
|
combos = list_combos()
|
|
tfs = all_tfs_for_combos(combos)
|
|
_set(symbols_total=len(symbols), running=True, last_error=None)
|
|
scanned = 0
|
|
errors = 0
|
|
changed_n = 0
|
|
|
|
for i, sym in enumerate(symbols):
|
|
try:
|
|
# Prefer low-TF of first combo for "enough history" gate
|
|
low0 = combos[0]["low"] if combos else "1d"
|
|
if bar_count(sym, low0) < 40:
|
|
backfill_symbol(sym, tfs)
|
|
tip_changed = tip_update_symbol(sym, tfs)
|
|
if tip_changed:
|
|
changed_n += 1
|
|
if force_rescan or tip_changed:
|
|
for combo in combos:
|
|
row = analyze_and_store(sym, combo_id=combo["id"])
|
|
if row:
|
|
scanned += 1
|
|
except Exception as e:
|
|
errors += 1
|
|
if errors <= 5:
|
|
logger.warning("tick %s: %s", sym, e)
|
|
_set(last_error=str(e))
|
|
if (i + 1) % 25 == 0:
|
|
_set(symbols_scanned=scanned)
|
|
logger.info("wyckoff tick progress %s/%s scanned=%s", i + 1, len(symbols), scanned)
|
|
|
|
_set(
|
|
running=False,
|
|
symbols_scanned=scanned,
|
|
last_tick_at=datetime.now(timezone.utc).isoformat(),
|
|
backfill_done=True,
|
|
)
|
|
return {
|
|
"symbols": len(symbols),
|
|
"scanned": scanned,
|
|
"changed_tips": changed_n,
|
|
"errors": errors,
|
|
"combos": [c["id"] for c in combos],
|
|
"tfs": tfs,
|
|
}
|
|
|
|
|
|
def _loop(interval: int, max_symbols: int | None):
|
|
try:
|
|
run_tick(max_symbols=max_symbols, force_rescan=True)
|
|
except Exception as e:
|
|
logger.exception("initial tick failed: %s", e)
|
|
_set(last_error=str(e), running=False)
|
|
while not _stop.wait(interval):
|
|
try:
|
|
# Tip-driven: only force full rescan when tips change is handled inside
|
|
run_tick(max_symbols=max_symbols, force_rescan=False)
|
|
except Exception as e:
|
|
logger.exception("tick failed: %s", e)
|
|
_set(last_error=str(e), running=False)
|
|
|
|
|
|
def start_scheduler(interval_sec: int = 60, max_symbols: int | None = None) -> None:
|
|
global _thread
|
|
if _thread and _thread.is_alive():
|
|
return
|
|
_stop.clear()
|
|
_set(tick_interval_sec=interval_sec)
|
|
_thread = threading.Thread(
|
|
target=_loop,
|
|
args=(interval_sec, max_symbols),
|
|
name="crypto-wyckoff-scheduler",
|
|
daemon=True,
|
|
)
|
|
_thread.start()
|
|
logger.info("crypto wyckoff scheduler started interval=%ss", interval_sec)
|
|
|
|
|
|
def stop_scheduler() -> None:
|
|
_stop.set()
|