# -*- coding: utf-8 -*- """合成K线数据生成器:供测试使用(无网络依赖)""" import numpy as np import pandas as pd def gen_kline(n=120, base=10.0, trend=0.001, vol_pct=0.02, seed=42): """ 生成合成日K线。 trend: 每日平均漂移(如 0.001 = +0.1%/天) vol_pct: 每日随机波动幅度 返回 DataFrame: [bar_time, open, high, low, close, volume, amount] """ rng = np.random.RandomState(seed) dates = pd.date_range('2025-01-01', periods=n, freq='B') close = np.zeros(n) close[0] = base for i in range(1, n): close[i] = close[i-1] * (1 + trend + rng.randn() * vol_pct) open_ = close * (1 + rng.randn(n) * vol_pct * 0.3) high = np.maximum(open_, close) * (1 + abs(rng.randn(n)) * vol_pct * 0.3) low = np.minimum(open_, close) * (1 - abs(rng.randn(n)) * vol_pct * 0.3) volume = np.abs(rng.randn(n)) * 1e6 + 5e5 amount = volume * close return pd.DataFrame({ 'bar_time': dates.strftime('%Y-%m-%d'), 'open': open_, 'high': high, 'low': low, 'close': close, 'volume': volume, 'amount': amount, }) def gen_trending_up(n=80, base=10.0): """持续上涨趋势K线""" return gen_kline(n=n, base=base, trend=0.008, vol_pct=0.01, seed=7) def gen_trending_down(n=80, base=50.0): """持续下跌趋势K线""" return gen_kline(n=n, base=base, trend=-0.008, vol_pct=0.01, seed=7)