104 lines
2.8 KiB
Python
104 lines
2.8 KiB
Python
"""生成测试数据以便在无法连接Binance时使用"""
|
|
|
|
import sys
|
|
import os
|
|
from pathlib import Path
|
|
|
|
# 添加项目根目录 to path
|
|
sys.path.append(str(Path(__file__).resolve().parent.parent))
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
|
|
from data.database import DatabaseManager
|
|
|
|
def generate_sample_data(
|
|
symbol: str = 'BTC',
|
|
timeframe: str = '1h',
|
|
num_days: int = 365
|
|
):
|
|
"""生成模拟的OHLCV数据用于测试
|
|
|
|
Args:
|
|
symbol: 币种符号
|
|
timeframe: 时间框架
|
|
num_days: 生成数据的天数
|
|
"""
|
|
# 计算数据点数量
|
|
if timeframe == '1h':
|
|
periods = num_days * 24
|
|
elif timeframe == '15m':
|
|
periods = num_days * 24 * 4
|
|
else:
|
|
periods = num_days * 24
|
|
|
|
# 生成时间戳
|
|
end_time = datetime.now()
|
|
start_time = end_time - timedelta(days=num_days)
|
|
|
|
if timeframe == '1h':
|
|
timestamps = pd.date_range(start=start_time, end=end_time, freq='h')
|
|
else: # 15m
|
|
timestamps = pd.date_range(start=start_time, end=end_time, freq='15min')
|
|
|
|
# 生成价格数据（随机游走）
|
|
base_price = {
|
|
'BTC': 45000,
|
|
'ETH': 2500,
|
|
'SOL': 100
|
|
}.get(symbol, 10000)
|
|
|
|
np.random.seed(42) # 固定种子以便复现
|
|
|
|
# 生成收益率
|
|
returns = np.random.normal(0.0001, 0.02, len(timestamps))
|
|
prices = base_price * np.exp(np.cumsum(returns))
|
|
|
|
# 生成OHLCV数据
|
|
data = []
|
|
for i, (ts, close) in enumerate(zip(timestamps, prices)):
|
|
# 生成高低价
|
|
volatility = close * 0.01
|
|
high = close + np.random.uniform(0, volatility)
|
|
low = close - np.random.uniform(0, volatility)
|
|
|
|
# 开盘价（前一个收盘价）
|
|
if i == 0:
|
|
open_price = close
|
|
else:
|
|
open_price = prices[i-1]
|
|
|
|
# 成交量
|
|
volume = np.random.uniform(1000, 10000)
|
|
|
|
data.append({
|
|
'timestamp': ts,
|
|
'open': open_price,
|
|
'high': high,
|
|
'low': low,
|
|
'close': close,
|
|
'volume': volume
|
|
})
|
|
|
|
df = pd.DataFrame(data)
|
|
|
|
# 保存到数据库
|
|
# 注意：我们这里假设DatabaseManager的默认路径在脚本运行时是相对于CWD的
|
|
# 如果从根目录运行，./data/cache/market_data.db 是正确的
|
|
db_manager = DatabaseManager()
|
|
db_manager.save_ohlcv(df, symbol, timeframe)
|
|
|
|
print(f"✓ 已生成 {len(df)} 行模拟数据并保存到数据库")
|
|
return df
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 生成BTC、ETH、SOL的测试数据
|
|
for symbol in ['BTC', 'ETH', 'SOL']:
|
|
for timeframe in ['1h', '15m']:
|
|
print(f"\n生成 {symbol} {timeframe} 数据...")
|
|
generate_sample_data(symbol, timeframe)
|
|
|
|
print("\n✓ 所有测试数据已生成！")
|