jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 精简仓库为 chanlun 核心与 web 分析，移除威科夫与遗留模块

Browse Source 
删除根目录旧 Chan 模块、策略、配置、文档及 wyckoff 相关代码；更新缠论 pipeline 与笔中枢计算；补充 research 研究与 web 测试。

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
jackyu66git
2026-08-27 01:05:12 +08:00
co-authored by Cursor
parent 5c10e35b76
commit 7f393b93ed
360 changed files with 140008 additions and 41167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/tests/fixtures/analyze_contract_keys.json
Vendored
+22
View File
@@ -0,0 +1,22 @@
{
"required": [
"bi_list",
"bi_zs_list",
"bsp_list",
"chan_macd",
"klc_fx_info",
"klc_list",
"klc_trend",
"kline_data",
"macd",
"seg_list",
"timezone",
"uncompleted_bi_list",
"uncompleted_seg_list",
"uncompleted_zs_list",
"zs_list"
],
"optional_when": {
"include_structure_zones": ["structure_zones"]
}
}
web/tests/helpers.py
+28
View File
@@ -0,0 +1,28 @@
"""测试辅助函数。"""
from __future__ import annotations
import numpy as np
import pandas as pd
def make_ohlcv(n: int = 400, seed: int = 42) -> pd.DataFrame:
rng = np.random.default_rng(seed)
dates = pd.date_range("2024-01-01", periods=n, freq="5min", tz="UTC")
rets = rng.normal(0, 0.002, size=n)
close = 100 * np.exp(np.cumsum(rets))
open_ = np.roll(close, 1)
open_[0] = close[0]
spread = np.abs(rng.normal(0, 0.0015, size=n)) * close
high = np.maximum(open_, close) + spread
low = np.minimum(open_, close) - spread
volume = rng.uniform(100, 1000, size=n)
return pd.DataFrame(
{
"date": dates,
"open": open_,
"high": high,
"low": low,
"close": close,
"volume": volume,
}
)
web/tests/test_analyze_contract.py
+5 -103
View File
@@ -1,4 +1,4 @@
"""ECR-002：加深 /api/analyze 相关契约 —— mock 行情 + analyze_chan 关键字段快照。"""
"""加深 /api/analyze 相关契约 —— mock 行情 + analyze_chan 关键字段快照。"""
from __future__ import annotations
import json
@@ -10,25 +10,21 @@ import pandas as pd
import pytest
ROOT = Path(__file__).resolve().parents[2]
WEB_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "web"))
sys.path.insert(0, str(WEB_ROOT))
from tests.generate_golden import make_ohlcv # noqa: E402
from tests.helpers import make_ohlcv # noqa: E402
_CONTRACT_DOC = json.loads(
(ROOT / "tests" / "fixtures" / "analyze_contract_keys.json").read_text(encoding="utf-8")
(Path(__file__).resolve().parent / "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 = {
@@ -74,7 +70,6 @@ def test_klines_recent_returns_tail_only():
from app import app
df = make_ohlcv(n=30)
# analyze 蓝图 star-import 后绑定在 api.analyze 命名空间
with patch("api.analyze.get_kl_data", return_value=df):
client = app.test_client()
resp = client.get(
@@ -88,7 +83,6 @@ def test_klines_recent_returns_tail_only():
assert isinstance(body.get("kline_data"), list)
assert len(body["kline_data"]) == 2
assert "bi_list" not in body
assert "wyckoff" not in body
def test_contract_keys_stable():
@@ -119,7 +113,6 @@ def test_serialize_chan_macd_shape():
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)
@@ -133,7 +126,6 @@ def test_analyze_http_contract_with_mocked_kl():
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(
@@ -149,93 +141,3 @@ def test_analyze_http_contract_with_mocked_kl():
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" in payload
for k in WYCKOFF_KEYS:
assert k in payload["wyckoff"], f"missing wyckoff key: {k}"
def test_analyze_http_wyckoff_can_opt_out():
"""include_wyckoff=0 时可显式跳过威科夫。"""
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": 0,
},
)
assert resp.status_code == 200, resp.data[:500]
payload = resp.get_json()
assert payload is not None and "wyckoff" not in payload
def test_analyze_http_wyckoff_for_three_timeframes():
"""主/次/次次均返回各自 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": "4h",
"element_timeframe": "2h",
"sub_sub_timeframe": "1h",
"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
assert "wyckoff" in payload
assert "element_wyckoff" in payload
assert "sub_sub_wyckoff" in payload
for key in ("wyckoff", "element_wyckoff", "sub_sub_wyckoff"):
for k in WYCKOFF_KEYS:
assert k in payload[key], f"missing {k} in {key}"
def test_analyze_http_wyckoff_skipped_when_elements_only():
"""elements_only=true 时不返回 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",
},
)
assert resp.status_code == 200, resp.data[:500]
payload = resp.get_json()
assert payload is not None
assert "wyckoff" not in payload
web/tests/test_cn_stock_data_fetch.py
+4 -5
View File
@@ -5,12 +5,11 @@ from pathlib import Path
import pytest
# 将项目根目录加入 sys.path，确保可以直接导入业务代码
ROOT_DIR = Path(__file__).resolve().parents[4]
if str(ROOT_DIR) not in sys.path:
sys.path.append(str(ROOT_DIR))
WEB_ROOT = Path(__file__).resolve().parents[1]
if str(WEB_ROOT) not in sys.path:
sys.path.insert(0, str(WEB_ROOT))
from user_data.Chan.web.cn_stock_data import ChinaStockData
from cn_stock_data import ChinaStockData
@pytest.mark.network
web/tests/test_wyckoff_crypto_routes.py
-123
View File
@@ -1,123 +0,0 @@
"""ECR-009: page/API smoke without requiring live provider during assert."""
from __future__ import annotations
import os
import sys
import pytest
# Ensure repo root + web on path like app.py
_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_WEB = os.path.join(_ROOT, "web")
for p in (_ROOT, _WEB):
if p not in sys.path:
sys.path.insert(0, p)
os.environ.setdefault("CRYPTO_WYCKOFF_DISABLE", "1")
@pytest.fixture()
def client():
from app import create_app
app = create_app()
app.config["TESTING"] = True
with app.test_client() as c:
yield c
def test_wyckoff_crypto_page_ok(client):
resp = client.get("/wyckoff_crypto")
assert resp.status_code == 200
assert b"Crypto Wyckoff Screener" in resp.data
assert b"fCombo" in resp.data
assert b"chartCanvas" in resp.data
def test_wyckoff_crypto_meta_ok(client):
resp = client.get("/api/wyckoff_crypto/meta")
assert resp.status_code == 200
data = resp.get_json()
assert "engine_version" in data
assert data.get("combo", {}).get("id") == "h8_4_1"
assert data["combo"]["low"] == "1h"
ids = {c["id"] for c in data.get("combos") or []}
assert "h8_4_1" in ids and "d_w_m" in ids
def test_wyckoff_crypto_scan_ok(client):
resp = client.get("/api/wyckoff_crypto/scan?limit=5&combo_id=h8_4_1")
assert resp.status_code == 200
data = resp.get_json()
assert "rows" in data
assert data.get("combo", {}).get("id") == "h8_4_1"
def test_wyckoff_crypto_klines_bad_request(client):
resp = client.get("/api/wyckoff_crypto/klines")
assert resp.status_code == 400
def test_wyckoff_crypto_klines_ok(client):
resp = client.get(
"/api/wyckoff_crypto/klines?symbol=BTC/USDT:USDT&tf=1h&limit=10&combo_id=h8_4_1"
)
assert resp.status_code == 200
data = resp.get_json()
assert "items" in data
assert data.get("tf") == "1h"
assert data.get("intraday") is True
if data["items"]:
assert "datetime" in data["items"][0]
assert "ts" in data["items"][0]
assert "T" in data["items"][0]["datetime"]
assert "+08:00" in data["items"][0]["datetime"]
def test_wyckoff_crypto_klines_bad_limit_ok(client):
resp = client.get(
"/api/wyckoff_crypto/klines?symbol=BTC/USDT:USDT&tf=1h&limit=abc&combo_id=h8_4_1"
)
assert resp.status_code == 200
def test_wyckoff_crypto_overlay_ok(client):
resp = client.get(
"/api/wyckoff_crypto/overlay?symbol=BTC/USDT:USDT&tf=1h&bars=60&combo_id=h8_4_1"
)
assert resp.status_code == 200
data = resp.get_json()
assert "phases" in data
assert "events" in data
def test_combos_add_and_list(client, tmp_path, monkeypatch):
from crypto_wyckoff import combos as cm
monkeypatch.setattr(cm, "_COMBOS_FILE", tmp_path / "combos.json")
monkeypatch.setattr(cm, "_cache", None)
resp = client.get("/api/wyckoff_crypto/combos")
assert resp.status_code == 200
assert len(resp.get_json()["combos"]) >= 2
bad = client.post(
"/api/wyckoff_crypto/combos",
json={"high": "1h", "mid": "4h", "low": "8h"},
)
assert bad.status_code == 400
ok = client.post(
"/api/wyckoff_crypto/combos",
json={"high": "12h", "mid": "4h", "low": "1h", "label": "12h/4h/1h"},
)
assert ok.status_code == 200
cid = ok.get_json()["combo"]["id"]
assert cid == "12h_4h_1h"
deleted = client.delete(f"/api/wyckoff_crypto/combos/{cid}")
assert deleted.status_code == 200
builtin = client.delete("/api/wyckoff_crypto/combos/h8_4_1")
assert builtin.status_code == 400