From ac845ccfd0ac40e8beef1f281917e8cec069c9f5 Mon Sep 17 00:00:00 2001 From: jackyu66git Date: Wed, 12 Nov 2025 23:59:11 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=9C=AC=E5=9C=B0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E6=BA=90=EF=BC=8C=E4=BB=A5=E5=90=8E=E5=B0=B1=E5=8F=AF?= =?UTF-8?q?=E4=BB=A5=E7=9B=B4=E6=8E=A5=E7=94=A8=E6=9C=AC=E5=9C=B0=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=BA=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .DS_Store | Bin 10244 -> 10244 bytes .gitignore | 1 + ChanHeng.py | 414 ++++++++++++++++++++++++++++++ ChanKLC.py | 4 +- ChanMACD.py | 506 +++++++++++++++++++------------------ TF_DF.py | 8 +- datasvc/Dockerfile | 9 + datasvc/README.md | 161 +++++++++--- datasvc/app/main.py | 407 +++++++++++++++++++++++++---- datasvc/docker-compose.yml | 4 +- datasvc/requirements.txt | 1 + web/app.py | 311 +++++++++++++++++------ web/requirements.txt | 1 + web/templates/index.html | 80 +++--- 14 files changed, 1461 insertions(+), 446 deletions(-) create mode 100644 ChanHeng.py diff --git a/.DS_Store b/.DS_Store index 5b7d94d5f27c3fcacc3954dbf0bf250ed828ef2c..7514a52c54331064b787c4097ce0c53bde707149 100644 GIT binary patch delta 81 zcmZn(XbG6$F8U^hRb&SoBgZOjtv3@HqW3?)EZ%uvRVoKl>ela!yI!?=0BFcUK) d!{mix5}R*{32|;_SNO%U`I*RVMz|n9BLLdB7_R^T delta 59 zcmZn(XbG6$jIU^hRb_GTV|ZOohR3jb$fWSo3YRATd2F(J;01=5?@6@Ia7mJvO` J2p8941ORlt6mkFn 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 @@ +