Chan/web/tests/test_analyze_contract.py

"""ECR-002：加深 /api/analyze 相关契约 —— mock 行情 + analyze_chan 关键字段快照。"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from unittest.mock import patch

import pandas as pd
import pytest

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "web"))

from tests.generate_golden import make_ohlcv  # noqa: E402


_CONTRACT_DOC = json.loads(
    (ROOT / "tests" / "fixtures" / "analyze_contract_keys.json").read_text(encoding="utf-8")
)
CONTRACT_KEYS = (
    _CONTRACT_DOC["required"]
    if isinstance(_CONTRACT_DOC, dict) and "required" in _CONTRACT_DOC
    else _CONTRACT_DOC
)
WYCKOFF_KEYS = (
    _CONTRACT_DOC.get("wyckoff_keys", [])
    if isinstance(_CONTRACT_DOC, dict)
    else []
)

# analyze_chan 直接返回的对象字段（未序列化前）
ANALYZE_CHAN_KEYS = {
    "klc_list",
    "klu_list",
    "bi_list",
    "seg_list",
    "zs_list",
    "bi_zs_list",
    "bsp_list",
    "klc_fx_info",
    "chan_macd",
    "ema52_dict",
}

CHAN_MACD_SERIALIZED_KEYS = {
    "seg_list",
    "unittf_list",
    "histset_list",
    "high_position_list",
    "high_empty_list",
    "low_position_list",
    "low_empty_list",
    "return_zero_list",
    "cross0_up_list",
    "cross0_down_list",
    "klu_list",
}


def test_analyze_route_registered():
    from app import app

    rules = {r.rule for r in app.url_map.iter_rules()}
    assert "/api/analyze" in rules
    assert "/api/chart_metadata" in rules
    assert "/" in rules
    assert "/chan_tv" in rules


def test_contract_keys_stable():
    assert "bi_list" in CONTRACT_KEYS and "seg_list" in CONTRACT_KEYS
    for k in ("kline_data", "macd", "zs_list", "bsp_list", "chan_macd"):
        assert k in CONTRACT_KEYS


def test_analyze_chan_keys_on_fixture():
    from services.runtime import add_indicators, analyze_chan

    df = add_indicators(make_ohlcv(400))
    result = analyze_chan(df, symbol="TEST/USDT:USDT", timeframe="5m")
    assert set(result.keys()) == ANALYZE_CHAN_KEYS
    assert isinstance(result["bi_list"], list)
    assert isinstance(result["seg_list"], list)
    assert isinstance(result["chan_macd"], dict)
    for k in ("seg_list", "unittf_list", "histset_list"):
        assert k in result["chan_macd"]


def test_serialize_chan_macd_shape():
    from pytz import timezone

    from services.runtime import add_indicators, analyze_chan, serialize_chan_macd_data

    df = add_indicators(make_ohlcv(200))
    result = analyze_chan(df)
    serialized = serialize_chan_macd_data(result["chan_macd"], timezone("Asia/Shanghai"))
    assert set(serialized.keys()) == CHAN_MACD_SERIALIZED_KEYS
    # JSON 可序列化
    json.dumps(serialized)


def test_analyze_http_contract_with_mocked_kl():
    """Flask 测试客户端：mock get_kl_data，断言响应含契约关键字段。"""
    from app import app
    from services.runtime import add_indicators

    df = add_indicators(make_ohlcv(300))
    df = df.copy()
    if "timestamp" not in df.columns:
        df["timestamp"] = (pd.to_datetime(df["date"]).astype("int64") // 10**6).astype("int64")

    # analyze 路由使用 `from services.runtime import *`，须 patch 其模块命名空间
    with patch("api.analyze.get_kl_data", return_value=df):
        client = app.test_client()
        resp = client.get(
            "/api/analyze",
            query_string={
                "symbol": "BTC/USDT:USDT",
                "timeframe": "5m",
                "timezone": "Asia/Shanghai",
            },
        )
    assert resp.status_code == 200, resp.data[:500]
    payload = resp.get_json()
    assert payload is not None and "error" not in payload
    missing = [k for k in CONTRACT_KEYS if k not in payload]
    assert not missing, f"missing contract keys: {missing}"
    assert "wyckoff" not in payload


def test_analyze_http_wyckoff_opt_in():
    """include_wyckoff=1 时响应含 wyckoff 约定键；默认不返回。"""
    from app import app
    from services.runtime import add_indicators

    df = add_indicators(make_ohlcv(300))
    df = df.copy()
    if "timestamp" not in df.columns:
        df["timestamp"] = (pd.to_datetime(df["date"]).astype("int64") // 10**6).astype("int64")

    with patch("api.analyze.get_kl_data", return_value=df):
        client = app.test_client()
        resp = client.get(
            "/api/analyze",
            query_string={
                "symbol": "BTC/USDT:USDT",
                "timeframe": "5m",
                "timezone": "Asia/Shanghai",
                "include_wyckoff": 1,
            },
        )
    assert resp.status_code == 200, resp.data[:500]
    payload = resp.get_json()
    assert payload is not None and "wyckoff" in payload
    w = payload["wyckoff"]
    for k in WYCKOFF_KEYS:
        assert k in w, f"missing wyckoff key: {k}"


def test_analyze_http_wyckoff_skipped_when_elements_only():
    """elements_only=true 时即使 include_wyckoff=1 也不返回 wyckoff。"""
    from app import app
    from services.runtime import add_indicators

    df = add_indicators(make_ohlcv(300))
    df = df.copy()
    if "timestamp" not in df.columns:
        df["timestamp"] = (pd.to_datetime(df["date"]).astype("int64") // 10**6).astype("int64")

    with patch("api.analyze.get_kl_data", return_value=df):
        client = app.test_client()
        resp = client.get(
            "/api/analyze",
            query_string={
                "symbol": "BTC/USDT:USDT",
                "timeframe": "5m",
                "element_timeframe": "1m",
                "timezone": "Asia/Shanghai",
                "elements_only": "true",
                "include_wyckoff": 1,
            },
        )
    assert resp.status_code == 200, resp.data[:500]
    payload = resp.get_json()
    assert payload is not None
    assert "wyckoff" not in payload