chan_glue.py

"""
chan_glue.py — wires the user's notebook-style modules together at runtime.

The original chan_engine.py / chan_multilevel.py were written as notebook cells:
chan_multilevel references `ChanAnalyzer` / `Signal` via a commented-out import.
Because both files use `from __future__ import annotations` and resolve names at
call time, we can simply inject the symbols into the module namespace — the Chan
analysis logic itself is left 100% untouched.

Also provides a small LRU-cached analyzer factory (the original chan_common.py
was A-share specific and is not used in the US version).
"""
from __future__ import annotations

import hashlib
from collections import OrderedDict

import pandas as pd

import chan_engine
import chan_multilevel

# ── inject cross-module symbols (replaces the commented `from chan_engine import …`) ──
chan_multilevel.ChanAnalyzer = chan_engine.ChanAnalyzer
chan_multilevel.Signal = chan_engine.Signal
chan_engine.PIVOT_MAX_EXTEND_SEGS = chan_engine.PIVOT_MAX_EXTEND_SEGS  # no-op, keeps linters calm

# Re-exports for app code
ChanAnalyzer = chan_engine.ChanAnalyzer
Signal = chan_engine.Signal
MultiLevelChan = chan_multilevel.MultiLevelChan
MultiLevelSignal = chan_multilevel.MultiLevelSignal
resample_weekly = chan_multilevel.resample_weekly
resample_monthly = chan_multilevel.resample_monthly
set_analyzer_factory = chan_multilevel.set_analyzer_factory

# ── cached analyzer factory ──────────────────────────────────────────────
_CACHE: "OrderedDict[str, chan_engine.ChanAnalyzer]" = OrderedDict()
_CACHE_MAX = 64


def _df_key(level: str, df: pd.DataFrame) -> str:
    n = len(df)
    if n == 0:
        return f"{level}-empty"
    last = str(df['date'].iloc[-1])
    first = str(df['date'].iloc[0])
    tail_close = float(df['close'].iloc[-1])
    raw = f"{level}|{n}|{first}|{last}|{tail_close:.6f}"
    return hashlib.md5(raw.encode()).hexdigest()


def cached_analyzer(level: str, df: pd.DataFrame):
    key = _df_key(level, df)
    if key in _CACHE:
        _CACHE.move_to_end(key)
        return _CACHE[key]
    an = chan_engine.ChanAnalyzer(df.reset_index(drop=True))
    _CACHE[key] = an
    while len(_CACHE) > _CACHE_MAX:
        _CACHE.popitem(last=False)
    return an


def install():
    """Install the cached analyzer factory into MultiLevelChan."""
    set_analyzer_factory(cached_analyzer)


install()