diff --git a/.DS_Store b/.DS_Store
index 5b7d94d..7514a52 100644
Binary files a/.DS_Store and b/.DS_Store differ
diff --git a/.gitignore b/.gitignore
index 0937f1f..15160ab 100644
--- a/.gitignore
+++ b/.gitignore
@@ -28,3 +28,4 @@ feature_meta
*.sqlite-wal
.DS_Store
交易记录/~$交易规则.docx
+/datasvc/data
diff --git a/ChanHeng.py b/ChanHeng.py
new file mode 100644
index 0000000..973837a
--- /dev/null
+++ b/ChanHeng.py
@@ -0,0 +1,414 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+"""
+使用 ccxt 获取币安交易所所有 `*/USDT` 交易对最新 100 根 1 小时 K 线数据，并筛选出长期横盘的币种。
+
+横盘判定基于以下三项指标（均可通过命令行参数调整）：
+1. 价格振幅占均价的比例（默认 ≤ 5%）
+2. 收盘价线性回归斜率占均价的比例（默认 ≤ 0.05%）
+3. 收盘价标准差占均价的比例（默认 ≤ 1.5%）
+
+满足以上全部条件的交易对会被视为长期横盘。
+"""
+
+import argparse
+import csv
+import logging
+import math
+import statistics
+import sys
+import time
+from dataclasses import dataclass
+from typing import Iterable, List, Optional, Sequence
+
+import ccxt
+
+# python ChanHeng.py --range-threshold 5 --slope-threshold 5 --std-threshold 0.015
+
+DEFAULT_LIMIT = 100
+DEFAULT_TIMEFRAME = "1h"
+STABLECOINS = {
+ "USDT",
+ "USDC",
+ "BUSD",
+ "TUSD",
+ "USDP",
+ "DAI",
+ "FDUSD",
+ "SUSD",
+ "UST",
+ "USTC",
+ "EUR",
+ "TRY",
+ "BFUSD",
+ "USDE",
+ "XUSD",
+ "USD1",
+ "XUSD"
+}
+
+
+@dataclass
+class SidewaysMetrics:
+ symbol: str
+ price_range_pct: float
+ slope_pct: float
+ std_pct: float
+ mean_close: float
+ last_close: float
+ data_points: int
+
+
+def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
+ parser = argparse.ArgumentParser(
+ description="筛选币安长期横盘币种（默认 500 根 1 小时 K 线）"
+ )
+ parser.add_argument(
+ "--timeframe",
+ default=DEFAULT_TIMEFRAME,
+ help="K 线周期（默认：1h）",
+ )
+ parser.add_argument(
+ "--limit",
+ type=int,
+ default=DEFAULT_LIMIT,
+ help="每个交易对获取的 K 线数量（默认：500）",
+ )
+ parser.add_argument(
+ "--range-threshold",
+ type=float,
+ default=0.05,
+ help="最大价格振幅占均价比例阈值（默认：0.05，表示 5%%）",
+ )
+ parser.add_argument(
+ "--slope-threshold",
+ type=float,
+ default=0.0005,
+ help="线性回归斜率占均价比例阈值（默认：0.0005，约 0.05%%）",
+ )
+ parser.add_argument(
+ "--std-threshold",
+ type=float,
+ default=0.015,
+ help="标准差占均价比例阈值（默认：0.015，表示 1.5%%）",
+ )
+ parser.add_argument(
+ "--quote",
+ action="append",
+ default=[],
+ help="只保留指定计价货币的交易对，可重复指定（示例：--quote USDT --quote FDUSD）",
+ )
+ parser.add_argument(
+ "--symbol",
+ action="append",
+ default=[],
+ help="仅检测指定交易对，可重复（不指定则遍历所有符合条件的现货交易对）",
+ )
+ parser.add_argument(
+ "--max-symbols",
+ type=int,
+ default=None,
+ help="限制最多检测的交易对数量（用于调试）",
+ )
+ parser.add_argument(
+ "--sleep",
+ type=float,
+ default=0.35,
+ help="请求失败后的基础重试等待秒数（默认：0.35）",
+ )
+ parser.add_argument(
+ "--retries",
+ type=int,
+ default=3,
+ help="单个交易对请求失败后的最大重试次数（默认：3）",
+ )
+ parser.add_argument(
+ "--include-inactive",
+ action="store_true",
+ help="包含已下架/不可交易的交易对（默认不包含）",
+ )
+ parser.add_argument(
+ "--export",
+ type=str,
+ default=None,
+ help="将筛选结果导出为 CSV 文件的路径",
+ )
+ parser.add_argument(
+ "--verbose",
+ action="store_true",
+ help="输出更详细的日志信息",
+ )
+ return parser.parse_args(argv)
+
+
+def setup_logging(verbose: bool) -> None:
+ level = logging.DEBUG if verbose else logging.INFO
+ logging.basicConfig(
+ level=level,
+ format="%(asctime)s [%(levelname)s] %(message)s",
+ datefmt="%Y-%m-%d %H:%M:%S",
+ )
+
+
+def create_exchange() -> ccxt.binance:
+ exchange = ccxt.binance({"enableRateLimit": True})
+ exchange.options["defaultType"] = "spot"
+ return exchange
+
+
+def iter_target_symbols(
+ exchange: ccxt.binance,
+ quotes: Sequence[str],
+ includes: Sequence[str],
+ include_inactive: bool,
+) -> List[str]:
+ markets = exchange.load_markets()
+ filtered = []
+
+ quote_set = {quote.upper() for quote in quotes}
+ include_set = {sym.upper() for sym in includes}
+
+ for symbol, meta in markets.items():
+ if not meta.get("spot", False):
+ continue
+ if not include_inactive and meta.get("active") is False:
+ continue
+
+ normalized_symbol = symbol.upper()
+
+ if include_set and normalized_symbol not in include_set:
+ continue
+
+ parts = symbol.split("/")
+ if len(parts) != 2:
+ continue
+
+ base_asset, quote_asset = parts[0].upper(), parts[1].upper()
+
+ target_quote = quote_set or {"USDT"}
+ if quote_asset not in target_quote:
+ continue
+
+ if base_asset in STABLECOINS:
+ continue
+
+ filtered.append(symbol)
+
+ filtered.sort()
+ logging.info(
+ "已筛选 %s 个目标交易对（quote 过滤：%s，专门列表：%s）",
+ len(filtered),
+ ",".join(sorted(quote_set or {"USDT"})),
+ ",".join(sorted(include_set)) or "无",
+ )
+ return filtered
+
+
+def fetch_ohlcv_with_retry(
+ exchange: ccxt.binance,
+ symbol: str,
+ timeframe: str,
+ limit: int,
+ retries: int,
+ base_sleep: float,
+) -> List[List[float]]:
+ attempt = 0
+ while True:
+ try:
+ return exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
+ except ccxt.RateLimitExceeded as exc:
+ wait_time = max(exchange.rateLimit / 1000.0 if exchange.rateLimit else 0, base_sleep)
+ logging.debug("触发限频，等待 %.2f 秒后重试 %s：%s", wait_time, symbol, exc)
+ time.sleep(wait_time)
+ except (ccxt.NetworkError, ccxt.ExchangeError) as exc:
+ attempt += 1
+ if attempt > retries:
+ logging.warning("多次获取失败，跳过 %s：%s", symbol, exc)
+ return []
+ wait_time = base_sleep * attempt
+ logging.debug("请求失败，等待 %.2f 秒后重试 %s（第 %d 次）：%s", wait_time, symbol, attempt, exc)
+ time.sleep(wait_time)
+
+
+def linear_regression_slope(values: Sequence[float]) -> float:
+ n = len(values)
+ if n < 2:
+ return 0.0
+ mean_x = (n - 1) / 2.0
+ mean_y = sum(values) / n
+ numerator = 0.0
+ denominator = 0.0
+ for idx, value in enumerate(values):
+ dx = idx - mean_x
+ numerator += dx * (value - mean_y)
+ denominator += dx * dx
+
+ if denominator == 0:
+ return 0.0
+ return numerator / denominator
+
+
+def compute_sideways_metrics(closes: Sequence[float], symbol: str) -> Optional[SidewaysMetrics]:
+ if not closes:
+ return None
+
+ mean_close = sum(closes) / len(closes)
+ if math.isclose(mean_close, 0.0):
+ return None
+
+ max_close = max(closes)
+ min_close = min(closes)
+ price_range_pct = (max_close - min_close) / mean_close
+
+ slope = linear_regression_slope(closes)
+ slope_pct = slope / mean_close
+
+ std_dev = statistics.pstdev(closes) if len(closes) > 1 else 0.0
+ std_pct = std_dev / mean_close
+
+ return SidewaysMetrics(
+ symbol=symbol,
+ price_range_pct=price_range_pct,
+ slope_pct=slope_pct,
+ std_pct=std_pct,
+ mean_close=mean_close,
+ last_close=closes[-1],
+ data_points=len(closes),
+ )
+
+
+def is_sideways(metrics: SidewaysMetrics, range_threshold: float, slope_threshold: float, std_threshold: float) -> bool:
+ return (
+ metrics.price_range_pct <= range_threshold
+ and abs(metrics.slope_pct) <= slope_threshold
+ and metrics.std_pct <= std_threshold
+ )
+
+
+def export_results(path: str, results: Sequence[SidewaysMetrics]) -> None:
+ fieldnames = [
+ "symbol",
+ "price_range_pct",
+ "slope_pct",
+ "std_pct",
+ "mean_close",
+ "last_close",
+ "data_points",
+ ]
+ with open(path, "w", newline="", encoding="utf-8") as fp:
+ writer = csv.DictWriter(fp, fieldnames=fieldnames)
+ writer.writeheader()
+ for item in results:
+ writer.writerow(
+ {
+ "symbol": item.symbol,
+ "price_range_pct": f"{item.price_range_pct:.6f}",
+ "slope_pct": f"{item.slope_pct:.6f}",
+ "std_pct": f"{item.std_pct:.6f}",
+ "mean_close": f"{item.mean_close:.8f}",
+ "last_close": f"{item.last_close:.8f}",
+ "data_points": item.data_points,
+ }
+ )
+ logging.info("结果已导出至 %s", path)
+
+
+def run(argv: Optional[Sequence[str]] = None) -> int:
+ args = parse_args(argv)
+ if not args.quote:
+ args.quote = ["USDT"]
+ setup_logging(args.verbose)
+
+ exchange = create_exchange()
+ symbols = iter_target_symbols(
+ exchange=exchange,
+ quotes=args.quote,
+ includes=args.symbol,
+ include_inactive=args.include_inactive,
+ )
+
+ if args.max_symbols is not None:
+ symbols = symbols[: args.max_symbols]
+ logging.info("出于调试目的，仅检测前 %d 个交易对。", len(symbols))
+
+ if not symbols:
+ logging.error("未找到任何满足条件的交易对，请检查过滤条件。")
+ return 1
+
+ sideways_results: List[SidewaysMetrics] = []
+ total = len(symbols)
+
+ for idx, symbol in enumerate(symbols, start=1):
+ logging.info("(%d/%d) 正在获取 %s 的 %s K 线（limit=%d）", idx, total, symbol, args.timeframe, args.limit)
+ ohlcv = fetch_ohlcv_with_retry(
+ exchange=exchange,
+ symbol=symbol,
+ timeframe=args.timeframe,
+ limit=args.limit,
+ retries=args.retries,
+ base_sleep=args.sleep,
+ )
+
+ if len(ohlcv) < max(100, args.limit // 2):
+ logging.debug("交易对 %s 返回数据不足（%d 根），跳过。", symbol, len(ohlcv))
+ continue
+
+ closes = [entry[4] for entry in ohlcv if entry[4] is not None]
+ metrics = compute_sideways_metrics(closes, symbol)
+ if not metrics:
+ continue
+
+ if is_sideways(metrics, args.range_threshold, args.slope_threshold, args.std_threshold):
+ sideways_results.append(metrics)
+ logging.info(
+ "识别为横盘：%s | 振幅 %.2f%% | 斜率 %.4f%% | 标准差 %.2f%%",
+ symbol,
+ metrics.price_range_pct * 100,
+ metrics.slope_pct * 100,
+ metrics.std_pct * 100,
+ )
+ else:
+ logging.debug(
+ "未满足条件：%s | 振幅 %.2f%% | 斜率 %.4f%% | 标准差 %.2f%%",
+ symbol,
+ metrics.price_range_pct * 100,
+ metrics.slope_pct * 100,
+ metrics.std_pct * 100,
+ )
+
+ if not sideways_results:
+ logging.warning("未检测到满足定义的长期横盘交易对。")
+ return 0
+
+ sideways_results.sort(key=lambda item: (item.price_range_pct, abs(item.slope_pct), item.std_pct))
+ print("=" * 88)
+ print(
+ f"共识别 {len(sideways_results)} 个长期横盘交易对（阈值：振幅≤{args.range_threshold:.2%}，"
+ f"斜率≤{args.slope_threshold:.2%}，标准差≤{args.std_threshold:.2%}）"
+ )
+ print("=" * 88)
+ header = f"{'Symbol':15s} {'Range%':>10s} {'Slope%':>10s} {'STD%':>10s} {'Mean':>14s} {'Last':>14s} {'Count':>6s}"
+ print(header)
+ print("-" * len(header))
+ for item in sideways_results:
+ print(
+ f"{item.symbol:15s}"
+ f" {item.price_range_pct * 100:10.4f}"
+ f" {item.slope_pct * 100:10.4f}"
+ f" {item.std_pct * 100:10.4f}"
+ f" {item.mean_close:14.8f}"
+ f" {item.last_close:14.8f}"
+ f" {item.data_points:6d}"
+ )
+
+ if args.export:
+ export_results(args.export, sideways_results)
+
+ logging.info("任务完成。")
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(run())
+
diff --git a/ChanKLC.py b/ChanKLC.py
index ea82943..df0c7ef 100644
--- a/ChanKLC.py
+++ b/ChanKLC.py
@@ -145,9 +145,9 @@ class ChanKLC():
def set_state(self, state):
self.state = state
def check_klu_included(self, klu):
- if self.high >= klu.high:
+ if self.high >= klu.high-10:
# high大于，low小于，左包含
- if self.low <= klu.low:
+ if self.low <= klu.low+10:
self.add_klu(klu=klu)
# gn>gn-1
if self.dir == Chan_KLINE_DIR.UP:
diff --git a/ChanMACD.py b/ChanMACD.py
index 0515096..42d2af6 100644
--- a/ChanMACD.py
+++ b/ChanMACD.py
@@ -5,246 +5,270 @@ from ChanMACDUnitTF import ChanMACDUnitTF
from ChanMACDHistSet import ChanMACDHistSet
class ChanMACD():
- def __init__(self, klu_list: list[ChanKLU]):
- self.klu_list = klu_list
- self.seg_list = []
- self.unittf_list = []
- self.histset_list = []
- # 状态标记列表
- self.high_position_list = [] # 高位列表
- self.high_empty_list = [] # 高位空列表
- self.return_zero_list = [] # 归零轴列表
- self.cross0_up_list = [] # 向上穿越零轴列表
- self.cross0_down_list = [] # 向下穿越零轴列表
- # 计算段 / UnitTF / HistSet 及状态标记
- self.cal_macd_state()
- def cal_macd_state(self):
- last_seg = None
- last_unittf = None
- last_histset = None
- last_klu = None
- for klu in self.klu_list:
- # initialise first histset
- if klu.macd == 0 and klu.signal == 0 and klu.macdhist == 0:
- continue
- if last_histset is None:
- if klu.macdhist > 0:
- last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
- self.histset_list.append(last_histset)
- else:
- last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(last_histset)
- else:
- # initialise first seg and unittf
- if last_seg is None:
- # create histset afterwards
- if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
- if klu.macdhist > 0:
- last_histset.add_klu(klu)
- else:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(histset)
- last_histset.set_next(histset)
- histset.set_pre(last_histset)
- last_histset.set_end_klu(last_klu)
- last_histset = histset
- else:
- if klu.macdhist < 0:
- last_histset.add_klu(klu)
- else:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
- self.histset_list.append(histset)
- last_histset.set_next(histset)
- histset.set_pre(last_histset)
- last_histset.set_end_klu(last_klu)
- last_histset = histset
- if last_klu.signal >= 0 and klu.signal < 0:
- last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
- self.unittf_list.append(last_unittf)
- last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.UNDER, last_unittf)
- self.seg_list.append(last_seg)
- elif last_klu.signal <= 0 and klu.signal > 0:
- last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
- self.unittf_list.append(last_unittf)
- last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.ABOVE, last_unittf)
- self.seg_list.append(last_seg)
- # after the first seg and unittf
- else:
- # create histset afterwards
- if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
- if klu.macdhist > 0:
- last_histset.add_klu(klu)
- else:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(histset)
- last_histset.set_next(histset)
- histset.set_pre(last_histset)
- last_histset.set_end_klu(last_klu)
- last_histset = histset
- if last_unittf:
- last_unittf.add_histset(last_histset)
- else:
- if klu.macdhist < 0:
- last_histset.add_klu(klu)
- else:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
- self.histset_list.append(histset)
- last_histset.set_next(histset)
- histset.set_pre(last_histset)
- last_histset.set_end_klu(last_klu)
- last_histset = histset
- if last_unittf:
- last_unittf.add_histset(last_histset)
- if last_klu.signal >= 0 and klu.signal < 0:
- last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_seg.set_end_klu(last_klu)
- seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.UNDER, unittf)
- self.seg_list.append(seg)
- last_seg.set_next(seg)
- last_seg = seg
- last_unittf = unittf
- elif last_klu.signal <= 0 and klu.signal > 0:
- last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_seg.set_end_klu(last_klu)
- seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.ABOVE, unittf)
- self.seg_list.append(seg)
- last_seg.set_next(seg)
- last_seg = seg
- last_unittf = unittf
- elif last_unittf.is_end and last_klu.macd < klu.macd and klu.macd > klu.signal:
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.NEAR0, last_histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_seg.add_unittf(unittf)
- last_unittf = unittf
- last_seg.add_klu(klu)
- else:
- if not last_unittf.is_end:
- last_unittf.add_klu(klu)
- last_seg.add_klu(klu)
- last_klu = klu
- klu.cal_macd_state()
- #print(klu.time, klu.macd_state, klu.continue_div, klu.separate_div, klu.macd, klu.signal, klu.macdhist, klu.ema24, klu.ema52, klu.close)
- return self.klu_list
- def cal_macd(self):
- last_seg = None
- last_unittf = None
- last_histset = None
- histset = None
- last_klu = None
- for klu in self.klu_list:
- klu.cal_macd_state()
- print(klu.time, klu.macd_state)
- # 1) 只有当 MACD 已可用(非 UNKNOWN)时，才开始初始化段/单元
- if last_seg is None:
- if klu.macd_state != Chan_MACD_STATE.UNKNOWN:
- # 初始化首个直方图集合（根据当前柱体正负）
- if klu.macdhist >= 0:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
- else:
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(histset)
- last_histset = histset
-
- # 初始化首段
- seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.signal >= 0 else Chan_MACDSEG_DIR.UNDER
- seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, seg_dir, last_unittf)
- self.seg_list.append(seg)
- last_seg = seg
+ def __init__(self, klu_list: list[ChanKLU]):
+ self.klu_list = klu_list
+ self.seg_list = []
+ self.unittf_list = []
+ self.histset_list = []
+ # 状态标记列表
+ self.high_position_list = [] # 高位列表
+ self.high_empty_list = [] # 高位空列表
+ self.return_zero_list = [] # 归零轴列表
+ self.cross0_up_list = [] # 向上穿越零轴列表
+ self.cross0_down_list = [] # 向下穿越零轴列表
+ # 计算段 / UnitTF / HistSet 及状态标记
+ self.cal_macd_state()
+ self.get_klu_sd_list()
+ def get_klu_sd(self):
+ if self.klu_list:
+ sd = self.klu_list[-1].separate_div
+ if sd > 1:
+ print(self.klu_list[-1].time, sd)
+ return True
+ return False
+ def get_klu_sd_list(self):
+ sd_list = []
+ if self.klu_list:
+ for klu in self.klu_list:
+ hist = klu.macdhist
+ singal = False
+ if klu.pre and klu.next:
+ if klu.signal > 0:
+ signal = klu.pre.signal > klu.signal and klu.next.signal < klu.signal
+ else:
+ signal = klu.pre.signal < klu.signal and klu.next.signal > klu.signal
+ sd = klu.separate_div
+ if sd > 1 and ((hist > 0 and hist < 20) or (hist < 0 and hist > -20)):
+ sd_list.append(klu.time)
+ print(klu.time, sd)
+ return sd_list
+ def cal_macd_state(self):
+ last_seg = None
+ last_unittf = None
+ last_histset = None
+ last_klu = None
+ for klu in self.klu_list:
+ # initialise first histset
+ if klu.macd == 0 and klu.signal == 0 and klu.macdhist == 0:
+ continue
+ if last_histset is None:
+ if klu.macdhist > 0:
+ last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
+ self.histset_list.append(last_histset)
+ else:
+ last_histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(last_histset)
+ else:
+ # initialise first seg and unittf
+ if last_seg is None:
+ # create histset afterwards
+ if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
+ if klu.macdhist > 0:
+ last_histset.add_klu(klu)
+ else:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(histset)
+ last_histset.set_next(histset)
+ histset.set_pre(last_histset)
+ last_histset.set_end_klu(last_klu)
+ last_histset = histset
+ else:
+ if klu.macdhist < 0:
+ last_histset.add_klu(klu)
+ else:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
+ self.histset_list.append(histset)
+ last_histset.set_next(histset)
+ histset.set_pre(last_histset)
+ last_histset.set_end_klu(last_klu)
+ last_histset = histset
+ if last_klu.signal >= 0 and klu.signal < 0:
+ last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
+ self.unittf_list.append(last_unittf)
+ last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.UNDER, last_unittf)
+ self.seg_list.append(last_seg)
+ elif last_klu.signal <= 0 and klu.signal > 0:
+ last_unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
+ self.unittf_list.append(last_unittf)
+ last_seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, Chan_MACDSEG_DIR.ABOVE, last_unittf)
+ self.seg_list.append(last_seg)
+ # after the first seg and unittf
+ else:
+ # create histset afterwards
+ if last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
+ if klu.macdhist > 0:
+ last_histset.add_klu(klu)
+ else:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(histset)
+ last_histset.set_next(histset)
+ histset.set_pre(last_histset)
+ last_histset.set_end_klu(last_klu)
+ last_histset = histset
+ if last_unittf:
+ last_unittf.add_histset(last_histset)
+ else:
+ if klu.macdhist < 0:
+ last_histset.add_klu(klu)
+ else:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE)
+ self.histset_list.append(histset)
+ last_histset.set_next(histset)
+ histset.set_pre(last_histset)
+ last_histset.set_end_klu(last_klu)
+ last_histset = histset
+ if last_unittf:
+ last_unittf.add_histset(last_histset)
+ if last_klu.signal >= 0 and klu.signal < 0:
+ last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.UNDER, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_seg.set_end_klu(last_klu)
+ seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.UNDER, unittf)
+ self.seg_list.append(seg)
+ last_seg.set_next(seg)
+ last_seg = seg
+ last_unittf = unittf
+ elif last_klu.signal <= 0 and klu.signal > 0:
+ last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.CROSS0, last_histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_seg.set_end_klu(last_klu)
+ seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, Chan_MACDSEG_DIR.ABOVE, unittf)
+ self.seg_list.append(seg)
+ last_seg.set_next(seg)
+ last_seg = seg
+ last_unittf = unittf
+ elif last_unittf.is_end and last_klu.macd < klu.macd and klu.macd > klu.signal:
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, Chan_MACDUNITTF_DIR.ABOVE, Chan_MACDUNITTF_TYPE.NEAR0, last_histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_seg.add_unittf(unittf)
+ last_unittf = unittf
+ last_seg.add_klu(klu)
+ else:
+ if not last_unittf.is_end:
+ last_unittf.add_klu(klu)
+ last_seg.add_klu(klu)
+ last_klu = klu
+ klu.cal_macd_state()
+ #print(klu.time, klu.macd_state, klu.continue_div, klu.separate_div, klu.macd, klu.signal, klu.macdhist, klu.ema24, klu.ema52, klu.close)
+ return self.klu_list
+ def cal_macd(self):
+ last_seg = None
+ last_unittf = None
+ last_histset = None
+ histset = None
+ last_klu = None
+ for klu in self.klu_list:
+ klu.cal_macd_state()
+ print(klu.time, klu.macd_state)
+ # 1) 只有当 MACD 已可用(非 UNKNOWN)时，才开始初始化段/单元
+ if last_seg is None:
+ if klu.macd_state != Chan_MACD_STATE.UNKNOWN:
+ # 初始化首个直方图集合（根据当前柱体正负）
+ if klu.macdhist >= 0:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.ABOVE)
+ else:
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, None, Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(histset)
+ last_histset = histset
+
+ # 初始化首段
+ seg_dir = Chan_MACDSEG_DIR.ABOVE if klu.signal >= 0 else Chan_MACDSEG_DIR.UNDER
+ seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, None, seg_dir, last_unittf)
+ self.seg_list.append(seg)
+ last_seg = seg
- # 初始化首个UnitTF
- unittf_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, unittf_dir, Chan_MACDUNITTF_TYPE.START, histset)
- self.unittf_list.append(unittf)
- last_unittf = unittf
- last_seg.add_unittf(unittf)
- # 未就绪则继续等下一根；已就绪亦已完成首个结构初始化，继续下一根
- last_klu = klu
- continue
- # 3) 直方图集合（基于当前 unittf）
- if klu.macdhist >= 0:
- if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
- last_histset.add_klu(klu)
- else:
- # 结束旧 histset（以前一根结束更合理）
- if last_histset and last_klu:
- last_histset.set_end_klu(last_klu)
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(histset)
- if last_histset:
- last_histset.set_next(histset)
- last_histset = histset
- if last_unittf:
- last_unittf.add_histset(histset)
- else:
- if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.UNDER:
- last_histset.add_klu(klu)
- else:
- # 结束旧 histset（以前一根结束更合理）
- if last_histset and last_klu:
- last_histset.set_end_klu(last_klu)
- histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
- self.histset_list.append(histset)
- if last_histset:
- last_histset.set_next(histset)
- last_histset = histset
- if last_unittf:
- last_unittf.add_histset(histset)
- # 2) 过零切段（使用KLU中的穿越状态）
- if (klu.macd_state == Chan_MACD_STATE.CROSS0_UP or
- klu.macd_state == Chan_MACD_STATE.CROSS0_DOWN):
- # 结束旧 unittf
- last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
- # 新的单位时间周期
- new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.CROSS0, histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_unittf = unittf
- # 收尾旧段
- last_seg.set_end_klu(last_klu)
- # 新段方向取反
- new_dir = Chan_MACDSEG_DIR.UNDER if last_seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else Chan_MACDSEG_DIR.ABOVE
- seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, new_dir, last_unittf)
- self.seg_list.append(seg)
- last_seg.set_next(seg)
- last_seg = seg
- last_seg.add_unittf(unittf)
- else:
- # 4) UnitTF 状态机：用黄线Signal的归零轴
- if last_klu.macd_state == Chan_MACD_STATE.NEAR0 and last_unittf.div_count > 1:
- #print(klu.time, klu.macd_state)
- if klu.macd_state == Chan_MACD_STATE.RZ_UP:
- last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
- new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_unittf = unittf
- last_seg.add_unittf(unittf)
- elif klu.macd_state == Chan_MACD_STATE.RZ_DOWN:
- last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
- new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
- unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
- self.unittf_list.append(unittf)
- last_unittf.set_next(unittf)
- last_unittf = unittf
- last_seg.add_unittf(unittf)
- else:
- last_unittf.add_klu(klu)
- last_seg.add_klu(klu)
- else:
- last_unittf.add_klu(klu)
- last_seg.add_klu(klu)
+ # 初始化首个UnitTF
+ unittf_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, None, unittf_dir, Chan_MACDUNITTF_TYPE.START, histset)
+ self.unittf_list.append(unittf)
+ last_unittf = unittf
+ last_seg.add_unittf(unittf)
+ # 未就绪则继续等下一根；已就绪亦已完成首个结构初始化，继续下一根
+ last_klu = klu
+ continue
+ # 3) 直方图集合（基于当前 unittf）
+ if klu.macdhist >= 0:
+ if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.ABOVE:
+ last_histset.add_klu(klu)
+ else:
+ # 结束旧 histset（以前一根结束更合理）
+ if last_histset and last_klu:
+ last_histset.set_end_klu(last_klu)
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(histset)
+ if last_histset:
+ last_histset.set_next(histset)
+ last_histset = histset
+ if last_unittf:
+ last_unittf.add_histset(histset)
+ else:
+ if last_histset and last_histset.histset_dir == Chan_MACDHISTSET_DIR.UNDER:
+ last_histset.add_klu(klu)
+ else:
+ # 结束旧 histset（以前一根结束更合理）
+ if last_histset and last_klu:
+ last_histset.set_end_klu(last_klu)
+ histset = ChanMACDHistSet(len(self.histset_list), klu.time, klu, last_histset, Chan_MACDHISTSET_DIR.ABOVE if klu.macdhist >= 0 else Chan_MACDHISTSET_DIR.UNDER)
+ self.histset_list.append(histset)
+ if last_histset:
+ last_histset.set_next(histset)
+ last_histset = histset
+ if last_unittf:
+ last_unittf.add_histset(histset)
+ # 2) 过零切段（使用KLU中的穿越状态）
+ if (klu.macd_state == Chan_MACD_STATE.CROSS0_UP or
+ klu.macd_state == Chan_MACD_STATE.CROSS0_DOWN):
+ # 结束旧 unittf
+ last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.CROSS0)
+ # 新的单位时间周期
+ new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.CROSS0, histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_unittf = unittf
+ # 收尾旧段
+ last_seg.set_end_klu(last_klu)
+ # 新段方向取反
+ new_dir = Chan_MACDSEG_DIR.UNDER if last_seg.seg_dir == Chan_MACDSEG_DIR.ABOVE else Chan_MACDSEG_DIR.ABOVE
+ seg = ChanMACDSeg(len(self.seg_list), klu.time, klu, last_seg, new_dir, last_unittf)
+ self.seg_list.append(seg)
+ last_seg.set_next(seg)
+ last_seg = seg
+ last_seg.add_unittf(unittf)
+ else:
+ # 4) UnitTF 状态机：用黄线Signal的归零轴
+ if last_klu.macd_state == Chan_MACD_STATE.NEAR0 and last_unittf.div_count > 1:
+ #print(klu.time, klu.macd_state)
+ if klu.macd_state == Chan_MACD_STATE.RZ_UP:
+ last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
+ new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_unittf = unittf
+ last_seg.add_unittf(unittf)
+ elif klu.macd_state == Chan_MACD_STATE.RZ_DOWN:
+ last_unittf.set_end_klu(last_klu, Chan_MACDUNITTF_TYPE.NEAR0)
+ new_dir = Chan_MACDUNITTF_DIR.ABOVE if klu.signal >= 0 else Chan_MACDUNITTF_DIR.UNDER
+ unittf = ChanMACDUnitTF(len(self.unittf_list), klu.time, klu, last_unittf, new_dir, Chan_MACDUNITTF_TYPE.NEAR0, histset)
+ self.unittf_list.append(unittf)
+ last_unittf.set_next(unittf)
+ last_unittf = unittf
+ last_seg.add_unittf(unittf)
+ else:
+ last_unittf.add_klu(klu)
+ last_seg.add_klu(klu)
+ else:
+ last_unittf.add_klu(klu)
+ last_seg.add_klu(klu)
- last_klu = klu
- last_histset.set_end_klu(last_klu)
- last_unittf.set_end_klu(last_klu, None)
- last_seg.set_end_klu(last_klu)
- return self.klu_list
\ No newline at end of file
+ last_klu = klu
+ last_histset.set_end_klu(last_klu)
+ last_unittf.set_end_klu(last_klu, None)
+ last_seg.set_end_klu(last_klu)
+ return self.klu_list
\ No newline at end of file
diff --git a/TF_DF.py b/TF_DF.py
index 1ce24d5..e99a62a 100644
--- a/TF_DF.py
+++ b/TF_DF.py
@@ -123,12 +123,12 @@ class TF_DF():
return klu_state_list
def check_fx(self, klc):
if klc.pre and klc.next:
- if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low and klc.next.klc_dir == Chan_KLINE_DIR.DOWN:
+ if klc.high > klc.pre.high and klc.high > klc.next.high and klc.low > klc.pre.low and klc.low > klc.next.low and klc.close > klc.next.close:
#if (klc.close > klc.ema52 or klc.next.close > klc.next.ema52) and klc.macd > 0 and klc.macd > klc.macdhist:
klc.set_fx(Chan_FX_TYPE.TOP)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "TOP")
return Chan_FX_TYPE.TOP
- elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high and klc.next.klc_dir == Chan_KLINE_DIR.UP:
+ elif klc.low < klc.pre.low and klc.low < klc.next.low and klc.high < klc.pre.high and klc.high < klc.next.high and klc.close < klc.next.close:
#if (klc.close < klc.ema52 or klc.next.close < klc.next.ema52) and klc.macd < 0 and klc.macd < klc.macdhist:
klc.set_fx(Chan_FX_TYPE.BOTTOM)
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time, klc.macd, klc.state, klc.fx, "BOTTOM")
@@ -1387,8 +1387,8 @@ class TF_DF():
if i >= 2:
self._detect_triple_pattern(klu_list[i-2], klu_list[i-1], klu)
- if klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
- print(klu.time, klu.pattern, klu.lower_shadow_ratio, klu.upper_shadow_ratio, klu.body_ratio, klu.lower_shadow_ratio/klu.body_ratio, klu.upper_shadow_ratio/klu.body_ratio)
+ #if klu.pattern != Chan_KLU_PATTERN.UNKNOWN:
+ #print(klu.time, klu.pattern, klu.lower_shadow_ratio, klu.upper_shadow_ratio, klu.body_ratio, klu.lower_shadow_ratio/klu.body_ratio, klu.upper_shadow_ratio/klu.body_ratio)
return klu_list
def _detect_single_reversal_pattern(self, klu):
diff --git a/datasvc/Dockerfile b/datasvc/Dockerfile
index 44f15fb..c21cce6 100644
--- a/datasvc/Dockerfile
+++ b/datasvc/Dockerfile
@@ -10,6 +10,15 @@ RUN pip install -r /app/requirements.txt
COPY app /app/app
+ENV DATA_DIR=/data \
+ EXCHANGE=binance \
+ SYMBOLS=BTC/USDT:USDT,ETH/USDT:USDT \
+ TIMEFRAMES=1m,1h,1d,1w,1M \
+ START_FROM=2022-01-01 \
+ POLL_FACTOR=0.5
+
+VOLUME ["/data"]
+
EXPOSE 9000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "9000"]
diff --git a/datasvc/README.md b/datasvc/README.md
index 7a40f33..58ac665 100644
--- a/datasvc/README.md
+++ b/datasvc/README.md
@@ -1,48 +1,141 @@
-# Local Data Service (REST + WebSocket)
+# Local Data Service（REST + WebSocket）
-一键部署、跨平台的本地行情数据服务。默认抓取 Binance 永续合约 `BTC/USDT:USDT, ETH/USDT:USDT` 的 `1m/5m/15m/1h` K 线，增量写入本地 Parquet 并通过 WebSocket 推送。
+本服务基于 FastAPI + ccxt，自动拉取交易所行情、写入本地 Parquet，同时提供 REST 和 WebSocket 数据访问。
+自带时间周期聚合能力：只需抓取 `1m / 1h / 1d / 1w / 1M` 等基础周期，即可自动生成 `2m/3m/.../30m`、`2h/3h/.../16h` 等衍生周期。
-## 快速开始（方式B：已安装 Docker）
+---
+## 1. 环境准备
+
+### 1.1 依赖
+- Python ≥ 3.10（本地运行方式需要）
+- `pip install -r requirements.txt`（包含 `fastapi`, `uvicorn`, `ccxt`, `pandas`, `pyarrow`, `technical` 等）
+- 或者直接使用仓库内的 `docker-compose.yml`
+
+### 1.2 关键环境变量
+| 变量 | 说明 | 默认 |
+| --- | --- | --- |
+| `DATA_DIR` | 本地 Parquet 存储目录 | `/data` |
+| `EXCHANGE` | 交易所标识（目前支持 binance） | `binance` |
+| `SYMBOLS` | 逗号分隔的交易对列表 | `BTC/USDT:USDT,ETH/USDT:USDT` |
+| `TIMEFRAMES` | 基础抓取周期，逗号分隔 | `1m,1h,1d,1w,1M` |
+| `START_FROM` | 首次启动回补的起始 UTC 时间（ISO 字符串或毫秒时间戳） | `2022-01-01` |
+| `POLL_FACTOR` | 拉取间隔因子，实际间隔 = 周期毫秒 × factor | `0.5` |
+| `BACKOFF_BASE / BACKOFF_MAX` | 异常重试的指数退避参数 | `2.0 / 30.0` |
+
+> 衍生周期列表由程序自动推导，无需手动写入 `TIMEFRAMES`。
+
+---
+
+## 2. 启动与关闭
+
+### 2.1 Docker 方式
```bash
cd user_data/Chan/datasvc
-docker compose up -d
+docker compose up -d # 启动
+docker compose logs -f # 查看日志
+docker compose down # 关闭
```
-- REST: http://localhost:9000/api/candles?symbol=BTC/USDT:USDT&tf=1m
-- WS: ws://localhost:9000/ws?symbol=BTC/USDT:USDT&tf=1m&since=1690000000000
-- Swagger: http://localhost:9000/docs
-
-## 环境变量（docker-compose.yml）
-- EXCHANGE: 交易所，默认 binance
-- SYMBOLS: 逗号分隔交易对
-- TIMEFRAMES: 逗号分隔周期
-- START_DAYS: 首次启动回补最近 N 天
-- POLL_FACTOR: 轮询因子，间隔=周期毫秒*factor
-- DATA_DIR: 容器内数据目录（已映射到 `./data`）
-
-## 数据位置
-- 本地缓存：`user_data/Chan/datasvc/data/{timeframe}/{symbol}.parquet`
-
-## 常用命令
+### 2.2 本地运行（无 Docker）
```bash
-docker compose logs -f
+export DATA_DIR=./data
+export SYMBOLS="BTC/USDT:USDT"
+export TIMEFRAMES="1m,1h,1d"
-docker compose down
+cd /Users/jack/Project/freqtrade
+uvicorn user_data.Chan.datasvc.app.main:app --reload
```
-## 接口说明
-- GET /api/candles
- - 参数：symbol, tf, start(ms), end(ms)
- - 返回：[{timestamp, open, high, low, close, volume}]
-- WS /ws
- - 参数：symbol, tf, since(ms)
- - 消息：
- - snapshot: 初始快照数组
- - upsert: 单根K线增量（尾部修正）
+关闭时 Ctrl+C 即可，服务会自动取消后台抓取任务并释放资源。
-## 注意
-- 默认未带交易所 API Key，仅公共行情。
-- 如需更多交易对/周期，修改 `docker-compose.yml` 后重启。
+---
+## 3. 数据存储与聚合
+### 3.1 基础周期
+只会为 `TIMEFRAMES` 声明的基础周期创建抓取任务（例如 `1m / 1h / 1d`）。
+
+### 3.2 衍生周期
+启动后自动维护以下聚合：
+
+| 基础周期 | 自动生成 |
+| --- | --- |
+| `1m` | `2m, 3m, 4m, 5m, 10m, 15m, 20m, 25m, 30m` |
+| `1h` | `2h, 3h, 4h, 6h, 8h, 12h, 16h` |
+| `1d` | `2d, 3d, 4d, 5d, 6d` |
+| `1w` | `2w` |
+| `1M` | `2M, 3M, 6M` |
+
+聚合过程通过 `technical.util.resample_to_interval` 完成，写入同一 Parquet 数据目录。
+所有周期都可以被 REST/WS 访问。
+
+### 3.3 数据目录
+```
+{DATA_DIR}/{timeframe}/{symbol}.parquet
+```
+
+---
+
+## 4. 接口调用
+
+### 4.1 健康检查
+```
+GET /health
+```
+返回运行状态、基础/衍生周期列表、各抓取任务的最新进度与错误计数，便于监控。
+
+### 4.2 REST API
+```
+GET /api/candles?symbol=BTC/USDT:USDT&tf=2h&start=1700000000000&end=1700003600000
+```
+参数说明：
+- `symbol`：交易对（必须在 `SYMBOLS` 列表中）
+- `tf`：时间周期（支持基础或衍生）
+- `start` / `end`：毫秒时间戳，可选
+
+返回示例：
+```json
+[
+ {"timestamp": 1700000000000, "open": 36000.0, "high": 36120.0, "low": 35980.0, "close": 36050.0, "volume": 125.4},
+ ...
+]
+```
+
+### 4.3 WebSocket
+```
+ws://localhost:8000/ws?symbol=ETH/USDT:USDT&tf=15m&since=1700000000000
+```
+- 首次连接：收到 `snapshot` 消息（快照数组）
+- 后续增量：收到 `upsert` 消息（最新几根K线），以及周期性 `ping`
+
+消息示例：
+```json
+{"topic":"candles.ETH/USDT:USDT.15m","type":"snapshot","data":[{"t":1700000000000,"o":2000.0,"h":2005.0,"l":1995.0,"c":2002.5,"v":312.7}, ...]}
+{"topic":"candles.ETH/USDT:USDT.15m","type":"upsert","data":{"t":1700000900000,"o":2002.5,"h":2006.0,"l":2000.0,"c":2004.0,"v":120.8}}
+```
+
+---
+
+## 5. 停机与维护
+
+- **正常关闭**：`docker compose down` 或 Ctrl+C。服务会等待所有抓取任务结束并关闭 `ccxt` 客户端。
+- **异常恢复**：若网络异常，服务会自动指数退避重试；可通过 `/health` 的 `consecutive_errors` 与 `last_error` 排查。
+- **数据清理**：直接删除 `DATA_DIR` 下对应的 Parquet 文件即可，下次启动会重新回补。
+
+---
+
+## 6. 常见问题
+
+1. **缺少 `technical` 模块**
+ 聚合周期会跳过，并在日志中提示；先执行 `pip install technical` 再重启。
+
+2. **接收不到某个周期的数据**
+ 确认该周期在 `TIMEFRAMES` 或自动聚合列表中；若是衍生周期，需要确保对应基础周期已在运行。
+
+3. **如何新增交易对/周期**
+ 修改环境变量或 docker-compose 配置后，重启服务即可；Parquet 文件会按需生成。
+
+---
+
+欢迎结合自身策略或可视化前端直接消费本地数据服务。若要集成到其他项目，可直接引用 `/api/candles` 的 JSON 响应或订阅 `/ws` 的实时推送。
diff --git a/datasvc/app/main.py b/datasvc/app/main.py
index 85a5c78..a3d114a 100644
--- a/datasvc/app/main.py
+++ b/datasvc/app/main.py
@@ -1,15 +1,24 @@
import os
import asyncio
import json
+import logging
+from contextlib import suppress
+from dataclasses import dataclass, field
from datetime import datetime, timedelta
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Tuple, Union
import ccxt
+import ccxt.async_support as ccxt_async
import pandas as pd
-from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
+from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException, status
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
+try:
+ from technical.util import resample_to_interval # type: ignore
+except ImportError: # pragma: no cover - 环境缺失依赖时自动降级
+ resample_to_interval = None # type: ignore
+
from .storage import (
ensure_storage,
read_candles,
@@ -18,12 +27,86 @@ from .storage import (
)
+LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()
+logging.basicConfig(
+ level=LOG_LEVEL,
+ format="%(asctime)s %(levelname)s [%(name)s] %(message)s",
+)
+logger = logging.getLogger("datasvc")
+RESAMPLE_AVAILABLE = resample_to_interval is not None
+RESAMPLE_WARNING_EMITTED = False
+
+AGGREGATION_PLAN: Dict[str, List[str]] = {
+ "1m": ["2m", "3m", "4m", "5m", "10m", "15m", "20m", "25m", "30m"],
+ "1h": ["2h", "3h", "4h", "6h", "8h", "12h", "16h"],
+ "1d": ["2d", "3d", "4d", "5d", "6d"],
+ "1w": ["2w"],
+ "1M": ["2M", "3M", "6M"],
+}
+
+CandleRow = List[Union[int, float]]
+
+
+def _split_env_list(value: str) -> List[str]:
+ return [item.strip() for item in value.split(",") if item.strip()]
+
+
+def _unique_preserve(values: List[str]) -> List[str]:
+ seen = set()
+ ordered: List[str] = []
+ for item in values:
+ if item not in seen:
+ ordered.append(item)
+ seen.add(item)
+ return ordered
+
+
+def timeframe_to_minutes(tf: str) -> Optional[int]:
+ if not tf:
+ return None
+ unit = tf[-1]
+ try:
+ value = int(tf[:-1])
+ except ValueError:
+ return None
+ multiplier = {
+ "m": 1,
+ "h": 60,
+ "d": 1440,
+ "w": 10080,
+ "M": 43200, # 30 天近似
+ }.get(unit)
+ if multiplier is None:
+ return None
+ return value * multiplier
+
DATA_DIR = os.environ.get("DATA_DIR", "/data")
EXCHANGE = os.environ.get("EXCHANGE", "binance")
-SYMBOLS = [s.strip() for s in os.environ.get("SYMBOLS", "BTC/USDT:USDT,ETH/USDT:USDT").split(",") if s.strip()]
-TIMEFRAMES = [t.strip() for t in os.environ.get("TIMEFRAMES", "1m,5m,15m,1h").split(",") if t.strip()]
-START_FROM = os.environ.get("START_FROM", "2025-01-01") # 首次启动拉取起始日期（UTC）
+SYMBOLS = _split_env_list(os.environ.get("SYMBOLS", "BTC/USDT:USDT,ETH/USDT:USDT"))
+if not SYMBOLS:
+ SYMBOLS = ["BTC/USDT:USDT"]
+
+_default_timeframes = ["1m", "1h", "1d", "1w", "1M"]
+requested_timeframes = _split_env_list(os.environ.get("TIMEFRAMES", ",".join(_default_timeframes)))
+if not requested_timeframes:
+ requested_timeframes = _default_timeframes
+
+FETCH_TIMEFRAMES = _unique_preserve(requested_timeframes)
+AVAILABLE_TIMEFRAMES = list(FETCH_TIMEFRAMES)
+for base_tf in FETCH_TIMEFRAMES:
+ for derived_tf in AGGREGATION_PLAN.get(base_tf, []):
+ if derived_tf not in AVAILABLE_TIMEFRAMES:
+ AVAILABLE_TIMEFRAMES.append(derived_tf)
+DERIVED_TIMEFRAMES = [tf for tf in AVAILABLE_TIMEFRAMES if tf not in FETCH_TIMEFRAMES]
+AGGREGATION_TARGETS = {tf: AGGREGATION_PLAN.get(tf, []) for tf in FETCH_TIMEFRAMES}
+
+START_FROM = os.environ.get("START_FROM", "2022-01-01") # 首次启动拉取起始日期（UTC）
POLL_FACTOR = float(os.environ.get("POLL_FACTOR", "0.5")) # 轮询间隔 = tf_ms * factor
+BACKOFF_BASE = float(os.environ.get("BACKOFF_BASE", "2.0"))
+BACKOFF_MAX = float(os.environ.get("BACKOFF_MAX", "30.0"))
+
+VALID_SYMBOLS = set(SYMBOLS)
+VALID_TIMEFRAMES = set(AVAILABLE_TIMEFRAMES)
ensure_storage(DATA_DIR)
@@ -38,18 +121,24 @@ app.add_middleware(
def tf_to_ms(tf: str) -> int:
- table = {
- "1m": 60_000,
- "3m": 3 * 60_000,
- "5m": 5 * 60_000,
- "15m": 15 * 60_000,
- "30m": 30 * 60_000,
- "1h": 60 * 60_000,
- "2h": 2 * 60 * 60_000,
- "4h": 4 * 60 * 60_000,
- "1d": 24 * 60 * 60_000,
- }
- return table.get(tf, 60_000)
+ minutes = timeframe_to_minutes(tf)
+ if minutes is None:
+ logger.warning("无法解析时间周期，默认使用 60 秒", extra={"timeframe": tf})
+ return 60_000
+ return minutes * 60_000
+
+
+def ensure_symbol_timeframe(symbol: str, timeframe: str) -> None:
+ if symbol not in VALID_SYMBOLS:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"symbol 必须为 {sorted(VALID_SYMBOLS)} 之一。",
+ )
+ if timeframe not in VALID_TIMEFRAMES:
+ raise HTTPException(
+ status_code=status.HTTP_400_BAD_REQUEST,
+ detail=f"tf 必须为 {sorted(VALID_TIMEFRAMES)} 之一。",
+ )
def parse_start_from_ms(val: str) -> int:
@@ -63,10 +152,10 @@ def parse_start_from_ms(val: str) -> int:
except Exception:
pass
try:
- dt = datetime.fromisoformat(val) # 允许 '2025-01-01' 或 '2025-01-01T00:00:00'
+ dt = datetime.fromisoformat(val) # 允许 '2022-01-01' 或 '2022-01-01T00:00:00'
except Exception:
# 回退到固定日期
- dt = datetime(2025, 1, 1)
+ dt = datetime(2022, 1, 1)
return int(dt.timestamp() * 1000)
@@ -104,52 +193,272 @@ class Hub:
hub = Hub()
+fetch_tasks: List[asyncio.Task] = []
+
+
+def resample_and_store(symbol: str, base_timeframe: str, derived_timeframes: List[str]) -> List[Tuple[str, List[CandleRow]]]:
+ if not RESAMPLE_AVAILABLE or not derived_timeframes:
+ return []
+ base_df = read_candles(DATA_DIR, symbol, base_timeframe, None, None)
+ if base_df.empty:
+ return []
+ base_df = base_df.copy()
+ if "date" not in base_df.columns:
+ base_df["date"] = pd.to_datetime(base_df["timestamp"], unit="ms", utc=True)
+ base_df = base_df.sort_values("timestamp")
+
+ updates: List[Tuple[str, List[List[float]]]] = []
+ for target_tf in derived_timeframes:
+ minutes = timeframe_to_minutes(target_tf)
+ if minutes is None:
+ logger.warning("无法解析聚合周期", extra={"target_timeframe": target_tf})
+ continue
+ try:
+ derived_df = resample_to_interval(base_df, minutes) # type: ignore[misc]
+ except Exception:
+ logger.exception(
+ "聚合周期计算失败",
+ extra={"symbol": symbol, "base_timeframe": base_timeframe, "target_timeframe": target_tf},
+ )
+ continue
+ if derived_df is None or derived_df.empty:
+ continue
+ derived_df = derived_df.copy()
+ if "timestamp" not in derived_df.columns:
+ if "date" in derived_df.columns:
+ dates = pd.to_datetime(derived_df["date"], utc=True, errors="coerce")
+ derived_df["timestamp"] = (dates.view("int64") // 1_000_000)
+ elif isinstance(derived_df.index, pd.DatetimeIndex):
+ idx = derived_df.index
+ if idx.tz is None:
+ idx = idx.tz_localize("UTC")
+ else:
+ idx = idx.tz_convert("UTC")
+ derived_df["timestamp"] = (idx.view("int64") // 1_000_000)
+ if "timestamp" not in derived_df.columns:
+ logger.warning(
+ "聚合结果缺少 timestamp 列，已跳过",
+ extra={"target_timeframe": target_tf},
+ )
+ continue
+ derived_df = derived_df.dropna(subset=["timestamp", "open", "high", "low", "close", "volume"])
+ if derived_df.empty:
+ continue
+ derived_df["timestamp"] = derived_df["timestamp"].astype("int64")
+ derived_df = derived_df.sort_values("timestamp")
+ last_ts = get_last_timestamp(DATA_DIR, symbol, target_tf)
+ if last_ts is not None:
+ derived_df = derived_df[derived_df["timestamp"] > last_ts]
+ if derived_df.empty:
+ continue
+ numpy_rows = derived_df[["timestamp", "open", "high", "low", "close", "volume"]].to_numpy()
+ records: List[CandleRow] = []
+ for ts, o, h, l, c, v in numpy_rows:
+ records.append(
+ [
+ int(ts),
+ float(o),
+ float(h),
+ float(l),
+ float(c),
+ float(v),
+ ]
+ )
+ if not records:
+ continue
+ upsert_candles(DATA_DIR, symbol, target_tf, records)
+ updates.append((target_tf, records[-3:] if len(records) > 3 else records))
+ return updates
+
+
+@dataclass
+class FetchState:
+ symbol: str
+ timeframe: str
+ started_at: datetime = field(default_factory=datetime.utcnow)
+ last_fetch_at: Optional[datetime] = None
+ last_candle_ts: Optional[int] = None
+ consecutive_errors: int = 0
+ last_error: Optional[str] = None
+
+ def to_payload(self) -> dict:
+ def serialize_dt(dt: Optional[datetime]) -> Optional[str]:
+ if not dt:
+ return None
+ return dt.replace(microsecond=0).isoformat() + "Z"
+
+ return {
+ "symbol": self.symbol,
+ "timeframe": self.timeframe,
+ "started_at": serialize_dt(self.started_at),
+ "last_fetch_at": serialize_dt(self.last_fetch_at),
+ "last_candle_ts": self.last_candle_ts,
+ "consecutive_errors": self.consecutive_errors,
+ "last_error": self.last_error,
+ }
+
+
+fetch_states: Dict[Tuple[str, str], FetchState] = {}
+
def build_exchange():
if EXCHANGE.lower() == "binance":
- return ccxt.binance({"enableRateLimit": True})
+ return ccxt_async.binance({"enableRateLimit": True})
raise RuntimeError(f"Unsupported EXCHANGE: {EXCHANGE}")
async def fetch_loop(symbol: str, timeframe: str):
"""持续增量抓取并广播。"""
+ derived_timeframes = AGGREGATION_TARGETS.get(timeframe, [])
+ global RESAMPLE_WARNING_EMITTED
+ if derived_timeframes and not RESAMPLE_AVAILABLE and not RESAMPLE_WARNING_EMITTED:
+ logger.warning(
+ "缺少 technical.util.resample_to_interval 模块，聚合时间周期生成已跳过",
+ extra={"timeframe": timeframe},
+ )
+ RESAMPLE_WARNING_EMITTED = True
+
exchange = build_exchange()
tf_ms = tf_to_ms(timeframe)
start_since = parse_start_from_ms(START_FROM)
last_ts = get_last_timestamp(DATA_DIR, symbol, timeframe)
since = max(start_since, (last_ts + tf_ms) if last_ts else start_since)
+ backoff = 1.0
+ state_key = (symbol, timeframe)
+ fetch_states[state_key] = FetchState(symbol=symbol, timeframe=timeframe, last_candle_ts=last_ts)
- while True:
- try:
- candles = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
- if candles:
- upsert_candles(DATA_DIR, symbol, timeframe, candles)
- for row in candles[-3:]:
- payload = {
- "topic": f"candles.{symbol}.{timeframe}",
- "type": "upsert",
- "data": {
- "t": row[0],
- "o": row[1],
- "h": row[2],
- "l": row[3],
- "c": row[4],
- "v": row[5],
- },
- }
- await hub.publish(symbol, timeframe, payload)
- since = candles[-1][0] + tf_ms
- await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0))
- except Exception:
- await asyncio.sleep(3.0)
+ logger.info("启动拉取任务", extra={"symbol": symbol, "timeframe": timeframe})
+
+ try:
+ while True:
+ try:
+ candles = await exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
+ if candles:
+ upsert_candles(DATA_DIR, symbol, timeframe, candles)
+ derived_updates: List[Tuple[str, List[CandleRow]]] = []
+ if derived_timeframes and RESAMPLE_AVAILABLE:
+ derived_updates = await asyncio.to_thread(
+ resample_and_store,
+ symbol,
+ timeframe,
+ derived_timeframes,
+ )
+ for row in candles[-3:]:
+ payload = {
+ "topic": f"candles.{symbol}.{timeframe}",
+ "type": "upsert",
+ "data": {
+ "t": row[0],
+ "o": row[1],
+ "h": row[2],
+ "l": row[3],
+ "c": row[4],
+ "v": row[5],
+ },
+ }
+ await hub.publish(symbol, timeframe, payload)
+ for target_tf, rows in derived_updates:
+ if not rows:
+ continue
+ for row in rows:
+ ts = int(row[0])
+ o, h, l, c, v = map(float, row[1:])
+ payload = {
+ "topic": f"candles.{symbol}.{target_tf}",
+ "type": "upsert",
+ "data": {
+ "t": ts,
+ "o": o,
+ "h": h,
+ "l": l,
+ "c": c,
+ "v": v,
+ },
+ }
+ await hub.publish(symbol, target_tf, payload)
+ since = candles[-1][0] + tf_ms
+ backoff = 1.0
+ state = fetch_states[state_key]
+ state.last_fetch_at = datetime.utcnow()
+ state.last_candle_ts = candles[-1][0]
+ state.consecutive_errors = 0
+ state.last_error = None
+ await asyncio.sleep(max(1.0, tf_ms * POLL_FACTOR / 1000.0))
+ except asyncio.CancelledError:
+ raise
+ except (ccxt.NetworkError, ccxt.ExchangeNotAvailable, ccxt.RequestTimeout) as exc:
+ logger.warning(
+ "网络异常，准备重试",
+ extra={"symbol": symbol, "timeframe": timeframe, "error": str(exc)},
+ )
+ state = fetch_states[state_key]
+ state.last_error = str(exc)
+ state.consecutive_errors += 1
+ backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
+ await asyncio.sleep(backoff)
+ except Exception as exc:
+ logger.exception(
+ "抓取循环发生异常，重建客户端后重试",
+ extra={"symbol": symbol, "timeframe": timeframe},
+ )
+ state = fetch_states[state_key]
+ state.last_error = str(exc)
+ state.consecutive_errors += 1
+ await asyncio.sleep(backoff)
+ with suppress(Exception):
+ await exchange.close()
+ exchange = build_exchange()
+ backoff = min(backoff * BACKOFF_BASE, BACKOFF_MAX)
+ except asyncio.CancelledError:
+ logger.info("取消拉取任务", extra={"symbol": symbol, "timeframe": timeframe})
+ state = fetch_states.get(state_key)
+ if state:
+ state.last_error = "cancelled"
+ raise
+ finally:
+ with suppress(Exception):
+ await exchange.close()
+ logger.info("拉取任务退出", extra={"symbol": symbol, "timeframe": timeframe})
@app.on_event("startup")
async def on_start():
ensure_storage(DATA_DIR)
+ fetch_tasks.clear()
for s in SYMBOLS:
- for tf in TIMEFRAMES:
- asyncio.create_task(fetch_loop(s, tf))
+ for tf in FETCH_TIMEFRAMES:
+ task = asyncio.create_task(fetch_loop(s, tf), name=f"fetch::{s}::{tf}")
+ fetch_tasks.append(task)
+
+
+@app.on_event("shutdown")
+async def on_shutdown():
+ if not fetch_tasks:
+ return
+ logger.info("正在停止拉取任务")
+ tasks = list(fetch_tasks)
+ for task in tasks:
+ task.cancel()
+ results = await asyncio.gather(*tasks, return_exceptions=True)
+ for result in results:
+ if isinstance(result, Exception) and not isinstance(result, asyncio.CancelledError):
+ logger.warning("任务停止时出现异常：%s", result)
+ fetch_tasks.clear()
+
+
+@app.get("/health")
+async def health():
+ now = datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+ return {
+ "status": "ok",
+ "time": now,
+ "exchange": EXCHANGE,
+ "symbols": SYMBOLS,
+ "base_timeframes": FETCH_TIMEFRAMES,
+ "derived_timeframes": DERIVED_TIMEFRAMES,
+ "timeframes": AVAILABLE_TIMEFRAMES,
+ "tasks": [state.to_payload() for state in fetch_states.values()],
+ }
@app.get("/api/candles")
@@ -160,6 +469,7 @@ def api_candles(
end: Optional[int] = Query(None, description="结束时间戳(ms)"),
):
try:
+ ensure_symbol_timeframe(symbol, tf)
df = read_candles(DATA_DIR, symbol, tf, start, end)
records = df.to_dict("records") if not df.empty else []
return JSONResponse(records)
@@ -169,6 +479,9 @@ def api_candles(
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket, symbol: str, tf: str, since: Optional[int] = None):
+ if symbol not in VALID_SYMBOLS or tf not in VALID_TIMEFRAMES:
+ await websocket.close(code=status.WS_1008_POLICY_VIOLATION, reason="invalid symbol/timeframe")
+ return
await hub.subscribe(websocket, symbol, tf)
try:
snap = read_candles(DATA_DIR, symbol, tf, since, None)
@@ -201,7 +514,9 @@ def root():
"service": "Local Data Service",
"exchange": EXCHANGE,
"symbols": SYMBOLS,
- "timeframes": TIMEFRAMES,
+ "base_timeframes": FETCH_TIMEFRAMES,
+ "derived_timeframes": DERIVED_TIMEFRAMES,
+ "timeframes": AVAILABLE_TIMEFRAMES,
}
diff --git a/datasvc/docker-compose.yml b/datasvc/docker-compose.yml
index bb586db..5dc5cea 100644
--- a/datasvc/docker-compose.yml
+++ b/datasvc/docker-compose.yml
@@ -7,8 +7,8 @@ services:
environment:
- EXCHANGE=binance
- SYMBOLS=BTC/USDT:USDT,ETH/USDT:USDT
- - TIMEFRAMES=1m,5m,15m,1h
- - START_FROM=2025-01-01
+ - TIMEFRAMES=1m,1h,1d,1w,1M
+ - START_FROM=2022-01-01
- POLL_FACTOR=0.5
- DATA_DIR=/data
- TZ=Asia/Shanghai
diff --git a/datasvc/requirements.txt b/datasvc/requirements.txt
index da6ce3b..ba66cd8 100644
--- a/datasvc/requirements.txt
+++ b/datasvc/requirements.txt
@@ -4,4 +4,5 @@ ccxt==4.4.27
pandas==2.2.2
pyarrow==16.1.0
orjson==3.10.3
+technical==1.5.0
diff --git a/web/app.py b/web/app.py
index a6cd768..5025552 100644
--- a/web/app.py
+++ b/web/app.py
@@ -1,6 +1,10 @@
from flask import Flask, render_template, jsonify, request
+from collections import OrderedDict
+import json
+import logging
import ccxt
import pandas as pd
+import requests
from datetime import datetime, timedelta
import sys
import os
@@ -41,31 +45,170 @@ exchange = ccxt.binance({
# 初始化A股数据获取器
china_stock = ChinaStockData()
-# 时间周期映射
-TIMEFRAMES = {
- '1m': '1分钟',
- '3m': '3分钟',
- '5m': '5分钟',
- '15m': '15分钟',
- '30m': '30分钟',
- '1h': '1小时',
- '2h': '2小时',
- '4h': '4小时',
- '6h': '6小时',
- '8h': '8小时',
- '12h': '12小时',
- '1d': '日线',
- '3d': '3日线',
- '1w': '周线',
- '1M': '月线',
-}
+logger = logging.getLogger(__name__)
-# 常见交易对
-SYMBOLS = [
+DATA_SERVICE_URL = os.environ.get("DATA_SERVICE_URL", os.environ.get("DATASVC_URL", "http://127.0.0.1:9000"))
+
+DEFAULT_TIMEFRAME_LABELS = OrderedDict([
+ ("1m", "1分钟"),
+ ("3m", "3分钟"),
+ ("5m", "5分钟"),
+ ("15m", "15分钟"),
+ ("30m", "30分钟"),
+ ("1h", "1小时"),
+ ("2h", "2小时"),
+ ("4h", "4小时"),
+ ("6h", "6小时"),
+ ("8h", "8小时"),
+ ("12h", "12小时"),
+ ("1d", "日线"),
+ ("3d", "3日线"),
+ ("1w", "周线"),
+ ("1M", "月线"),
+])
+
+DEFAULT_SYMBOLS = [
'SOL/USDT:USDT', 'BTC/USDT:USDT', 'ETH/USDT:USDT', 'BNB/USDT:USDT', 'XRP/USDT:USDT',
'ADA/USDT:USDT', 'DOGE/USDT:USDT', 'AVAX/USDT:USDT', 'DOT/USDT:USDT', 'MATIC/USDT:USDT'
]
+TIMEFRAMES = DEFAULT_TIMEFRAME_LABELS.copy()
+SYMBOLS = DEFAULT_SYMBOLS.copy()
+DATA_SERVICE_AVAILABLE = False
+SERVICE_METADATA_LAST_REFRESH = 0
+
+
+def timeframe_to_minutes(tf: str):
+ """将时间周期转换为分钟数，用于排序。"""
+ if not tf:
+ return None
+ unit = tf[-1]
+ try:
+ value = int(tf[:-1])
+ except (ValueError, TypeError):
+ return None
+ multiplier = {
+ 'm': 1,
+ 'h': 60,
+ 'd': 1440,
+ 'w': 10080,
+ 'M': 43200, # 30天近似
+ }.get(unit)
+ if multiplier is None:
+ return None
+ return value * multiplier
+
+
+def format_timeframe_label(tf: str) -> str:
+ """将时间周期转换为可读标签。"""
+ if not tf:
+ return tf
+ unit = tf[-1]
+ try:
+ value = int(tf[:-1])
+ except (ValueError, TypeError):
+ return tf
+ if unit == 'm':
+ return f"{value}分钟"
+ if unit == 'h':
+ return f"{value}小时"
+ if unit == 'd':
+ return "日线" if value == 1 else f"{value}日线"
+ if unit == 'w':
+ return "周线" if value == 1 else f"{value}周线"
+ if unit == 'M':
+ return "月线" if value == 1 else f"{value}月线"
+ return tf
+
+
+def build_timeframe_labels(timeframes):
+ ordered = sorted(
+ timeframes,
+ key=lambda tf: timeframe_to_minutes(tf) if timeframe_to_minutes(tf) is not None else float('inf'),
+ )
+ labels = OrderedDict()
+ for tf in ordered:
+ labels[tf] = format_timeframe_label(tf)
+ return labels
+
+
+def _parse_time_input(value):
+ if value in (None, '', 0):
+ return None
+ try:
+ return int(float(value))
+ except (ValueError, TypeError):
+ return None
+
+
+def refresh_data_service_metadata(force=False):
+ """刷新数据服务提供的交易对与周期元信息。"""
+ global DATA_SERVICE_AVAILABLE, TIMEFRAMES, SYMBOLS, SERVICE_METADATA_LAST_REFRESH
+ now = time.time()
+ if not force and DATA_SERVICE_AVAILABLE and now - SERVICE_METADATA_LAST_REFRESH < 60:
+ return True
+ try:
+ resp = requests.get(f"{DATA_SERVICE_URL}/health", timeout=5)
+ resp.raise_for_status()
+ payload = resp.json()
+ service_symbols = payload.get("symbols") or payload.get("symbol_list") or []
+ base_timeframes = payload.get("timeframes") or payload.get("base_timeframes") or []
+ derived = payload.get("derived_timeframes") or []
+ service_timeframes = list(base_timeframes)
+ for tf in derived:
+ if tf not in service_timeframes:
+ service_timeframes.append(tf)
+ if service_symbols:
+ SYMBOLS[:] = service_symbols
+ if service_timeframes:
+ TIMEFRAMES.clear()
+ TIMEFRAMES.update(build_timeframe_labels(service_timeframes))
+ DATA_SERVICE_AVAILABLE = True
+ SERVICE_METADATA_LAST_REFRESH = now
+ return True
+ except Exception as exc:
+ logger.warning("无法加载数据服务元信息: %s", exc)
+ if not DATA_SERVICE_AVAILABLE:
+ TIMEFRAMES.clear()
+ TIMEFRAMES.update(DEFAULT_TIMEFRAME_LABELS)
+ SYMBOLS[:] = DEFAULT_SYMBOLS
+ DATA_SERVICE_AVAILABLE = False
+ return False
+
+
+def _fetch_kl_from_datasvc(symbol, timeframe, start_ms=None, end_ms=None, limit=None):
+ params = {"symbol": symbol, "tf": timeframe}
+ if start_ms is not None:
+ params["start"] = int(start_ms)
+ if end_ms is not None:
+ params["end"] = int(end_ms)
+ resp = requests.get(f"{DATA_SERVICE_URL}/api/candles", params=params, timeout=10)
+ resp.raise_for_status()
+ data = resp.json()
+ if not data:
+ return None
+ df = pd.DataFrame(data)
+ if df.empty or "timestamp" not in df.columns:
+ return None
+ numeric_cols = ["open", "high", "low", "close", "volume"]
+ df["timestamp"] = pd.to_numeric(df["timestamp"], errors="coerce")
+ df = df.dropna(subset=["timestamp"])
+ df["timestamp"] = df["timestamp"].astype("int64")
+ for col in numeric_cols:
+ if col in df.columns:
+ df[col] = pd.to_numeric(df[col], errors="coerce")
+ df = df.dropna(subset=numeric_cols)
+ df = df.sort_values("timestamp")
+ if limit and len(df) > limit:
+ df = df.tail(limit)
+ df = df.reset_index(drop=True)
+ df["date"] = pd.to_datetime(df["timestamp"], unit='ms', utc=True).dt.tz_convert('Asia/Shanghai')
+ return df
+
+
+# 模块加载时尝试预取一次元信息，但失败不阻塞后续流程
+refresh_data_service_metadata(force=True)
+
# A股热门股票
A_STOCK_SYMBOLS = china_stock.get_popular_stocks()
@@ -89,7 +232,7 @@ def get_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
else:
return None
-def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
+def _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit=1000, start_time=None, end_time=None):
"""获取加密货币K线数据，支持分页加载确保获取指定时间范围内的所有数据"""
try:
# 初始化参数
@@ -202,6 +345,30 @@ def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=
except Exception as e:
return None
+
+def get_crypto_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
+ """优先通过本地数据服务获取加密货币K线，失败时回退至交易所API。"""
+ start_ms = _parse_time_input(start_time)
+ end_ms = _parse_time_input(end_time)
+
+ refresh_data_service_metadata()
+ if DATA_SERVICE_AVAILABLE:
+ try:
+ df = _fetch_kl_from_datasvc(
+ symbol=symbol,
+ timeframe=timeframe,
+ start_ms=start_ms,
+ end_ms=end_ms,
+ limit=limit,
+ )
+ if df is not None and not df.empty:
+ return df
+ except Exception as exc:
+ logger.warning("数据服务请求失败，准备回退至交易所 API：%s", exc)
+
+ return _get_crypto_kl_data_via_ccxt(symbol, timeframe, limit, start_time, end_time)
+
+
def get_a_stock_kl_data(symbol, timeframe, limit=1000, start_time=None, end_time=None):
"""获取A股K线数据"""
try:
@@ -733,66 +900,18 @@ def serialize_chan_macd_data(chan_macd_data, client_tz):
def is_smaller_timeframe(tf1, tf2):
"""判断时间周期tf1是否小于tf2"""
- # 定义时间周期的分钟数映射
- tf_values = {
- '1m': 1,
- '3m': 3,
- '5m': 5,
- '15m': 15,
- '30m': 30,
- '1h': 60,
- '2h': 120,
- '4h': 240,
- '6h': 360,
- '8h': 480,
- '12h': 720,
- '1d': 1440,
- '3d': 4320,
- '1w': 10080,
- '1M': 43200
- }
-
- # 获取时间周期对应的分钟数
- tf1_value = tf_values.get(tf1)
- tf2_value = tf_values.get(tf2)
-
- # 如果某个时间周期不在映射中，返回False
+ tf1_value = timeframe_to_minutes(tf1)
+ tf2_value = timeframe_to_minutes(tf2)
if tf1_value is None or tf2_value is None:
return False
-
- # 返回tf1是否小于tf2
return tf1_value < tf2_value
def is_smaller_or_equal_timeframe(tf1, tf2):
"""判断时间周期tf1是否小于等于tf2"""
- # 定义时间周期的分钟数映射
- tf_values = {
- '1m': 1,
- '3m': 3,
- '5m': 5,
- '15m': 15,
- '30m': 30,
- '1h': 60,
- '2h': 120,
- '4h': 240,
- '6h': 360,
- '8h': 480,
- '12h': 720,
- '1d': 1440,
- '3d': 4320,
- '1w': 10080,
- '1M': 43200
- }
-
- # 获取时间周期对应的分钟数
- tf1_value = tf_values.get(tf1)
- tf2_value = tf_values.get(tf2)
-
- # 如果某个时间周期不在映射中，返回False
+ tf1_value = timeframe_to_minutes(tf1)
+ tf2_value = timeframe_to_minutes(tf2)
if tf1_value is None or tf2_value is None:
return False
-
- # 返回tf1是否小于等于tf2
return tf1_value <= tf2_value
def clean_dataframe_for_json(df):
@@ -900,12 +1019,15 @@ def classify_trend_stage(df):
def load_crypto_symbols(limit=200):
"""加载常见USDT永续合约交易对，返回列表"""
+ refresh_data_service_metadata()
+ if SYMBOLS:
+ return SYMBOLS[:limit]
try:
markets = exchange.load_markets()
symbols = [s for s in markets.keys() if '/USDT' in s and ':USDT' in s]
return symbols[:limit]
except Exception:
- return SYMBOLS
+ return DEFAULT_SYMBOLS[:limit]
@app.route('/api/trend_filter', methods=['GET'])
@@ -1030,10 +1152,38 @@ def trend_detail():
@app.route('/')
def index():
"""主页"""
- return render_template('index.html',
- timeframes=TIMEFRAMES,
- symbols=SYMBOLS,
- a_stock_symbols=A_STOCK_SYMBOLS)
+ refresh_data_service_metadata()
+ timeframe_items = list(TIMEFRAMES.items())
+ timeframe_keys = [item[0] for item in timeframe_items]
+ symbols = SYMBOLS if SYMBOLS else DEFAULT_SYMBOLS
+
+ preferred_main = next((tf for tf in ['5m', '15m', '1h'] if tf in TIMEFRAMES), None)
+ default_main = preferred_main or (timeframe_keys[0] if timeframe_keys else '1m')
+ if default_main not in TIMEFRAMES and timeframe_keys:
+ default_main = timeframe_keys[0]
+
+ if timeframe_keys:
+ try:
+ idx = timeframe_keys.index(default_main)
+ default_element = timeframe_keys[idx - 1] if idx > 0 else timeframe_keys[0]
+ except ValueError:
+ default_element = timeframe_keys[0]
+ else:
+ default_element = default_main
+
+ default_symbol = 'BTC/USDT:USDT' if 'BTC/USDT:USDT' in symbols else (symbols[0] if symbols else '')
+
+ return render_template(
+ 'index.html',
+ timeframes=TIMEFRAMES,
+ symbols=symbols,
+ a_stock_symbols=A_STOCK_SYMBOLS,
+ default_main_timeframe=default_main,
+ default_element_timeframe=default_element,
+ default_symbol=default_symbol,
+ timeframe_keys_json=json.dumps(timeframe_keys),
+ data_service_available=DATA_SERVICE_AVAILABLE,
+ )
@app.route('/api/analyze')
def analyze():
@@ -1417,13 +1567,16 @@ def analyze():
@app.route('/api/symbols')
def get_symbols():
"""获取可用交易对"""
+ refresh_data_service_metadata()
+ if SYMBOLS:
+ return jsonify(SYMBOLS)
try:
markets = exchange.load_markets()
# 合约交易对通常是以USDT结尾的永续合约
symbols = [symbol for symbol in markets.keys() if '/USDT' in symbol and ':USDT' in symbol]
return jsonify(symbols)
except Exception as e:
- return jsonify({'error': str(e)})
+ return jsonify(DEFAULT_SYMBOLS)
@app.route('/api/a_stocks')
def get_a_stocks():
diff --git a/web/requirements.txt b/web/requirements.txt
index cc585d0..968c6b0 100644
--- a/web/requirements.txt
+++ b/web/requirements.txt
@@ -2,6 +2,7 @@ flask>=2.0.1
ccxt>=4.4.70
pandas>=1.3.3
numpy>=1.21.2
+requests>=2.31.0
plotly>=5.3.1
matplotlib>=3.4.3
pytz>=2021.1
diff --git a/web/templates/index.html b/web/templates/index.html
index 0d20d13..ede1d64 100644
--- a/web/templates/index.html
+++ b/web/templates/index.html
@@ -16,6 +16,25 @@
+