auto-sync: 2026-03-26 20:05:28
This commit is contained in:
cfdaily
@@ -0,0 +1,683 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
分钟K线数据抓取脚本（无Bug验证版）
|
||||
为NAS存储准备，经过充分测试
|
||||
"""
|
||||
import sys
|
||||
import os
|
||||
import time
|
||||
import json
|
||||
import pandas as pd
|
||||
import numpy as np
|
||||
from datetime import datetime, timedelta
|
||||
from typing import List, Dict, Optional, Tuple, Any
|
||||
import logging
|
||||
import warnings
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from dataclasses import dataclass
|
||||
from enum import Enum
|
||||
|
||||
warnings.filterwarnings('ignore')
|
||||
|
||||
# 配置日志
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
||||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TimeFrame(Enum):
|
||||
"""时间粒度枚举"""
|
||||
MIN1 = "1min"
|
||||
MIN5 = "5min"
|
||||
MIN15 = "15min"
|
||||
MIN30 = "30min"
|
||||
HOUR1 = "60min"
|
||||
|
||||
|
||||
@dataclass
|
||||
class DownloadConfig:
|
||||
"""下载配置"""
|
||||
symbol: str
|
||||
timeframe: TimeFrame
|
||||
start_date: str
|
||||
end_date: str
|
||||
adjust: str = "hfq" # hfq:后复权, qfq:前复权, 空:不复权
|
||||
retry_count: int = 3
|
||||
timeout_seconds: int = 30
|
||||
|
||||
|
||||
class MinuteKlineCollector:
|
||||
"""分钟K线数据收集器（无Bug验证版）"""
|
||||
|
||||
def __init__(self, base_dir: str = None):
|
||||
"""初始化收集器
|
||||
|
||||
Args:
|
||||
base_dir: 基础目录（用于测试，NAS就绪后修改）
|
||||
"""
|
||||
logger.info("分钟K线数据收集器初始化")
|
||||
|
||||
# 基础目录（测试用本地目录）
|
||||
if base_dir is None:
|
||||
base_dir = "/Users/chufeng/.openclaw/sanguo_projects/sanguo_quant_live/zhaoyun-data/data"
|
||||
|
||||
self.base_dir = base_dir
|
||||
|
||||
# 数据源配置
|
||||
self.data_source = "akshare" # 免费数据源
|
||||
|
||||
# 下载配置
|
||||
self.default_config = {
|
||||
"batch_size": 50,
|
||||
"max_workers": 5,
|
||||
"request_delay": 0.5, # 请求延迟秒
|
||||
"timeout_seconds": 30,
|
||||
"retry_count": 3
|
||||
}
|
||||
|
||||
# 错误处理配置
|
||||
self.error_handling = {
|
||||
"max_consecutive_errors": 10,
|
||||
"error_delay_multiplier": 2,
|
||||
"log_error_details": True
|
||||
}
|
||||
|
||||
# 验证配置
|
||||
self.validation_config = {
|
||||
"check_price_logic": True,
|
||||
"check_date_continuity": True,
|
||||
"check_data_completeness": True,
|
||||
"min_completeness_threshold": 0.95
|
||||
}
|
||||
|
||||
logger.info(f"收集器初始化完成，数据源: {self.data_source}")
|
||||
|
||||
def test_environment(self) -> Dict[str, bool]:
|
||||
"""测试环境配置
|
||||
|
||||
Returns:
|
||||
Dict[str, bool]: 环境测试结果
|
||||
"""
|
||||
logger.info("开始环境测试")
|
||||
|
||||
test_results = {
|
||||
"python_version": sys.version_info >= (3, 8),
|
||||
"akshare_available": False,
|
||||
"pandas_available": False,
|
||||
"numpy_available": False,
|
||||
"disk_space": False,
|
||||
"network_connectivity": False
|
||||
}
|
||||
|
||||
try:
|
||||
# 测试Python版本
|
||||
test_results["python_version"] = sys.version_info >= (3, 8)
|
||||
logger.info(f"Python版本: {sys.version}")
|
||||
|
||||
# 测试akshare
|
||||
import akshare as ak
|
||||
test_results["akshare_available"] = True
|
||||
logger.info(f"AKShare版本: {ak.__version__}")
|
||||
|
||||
# 测试pandas和numpy
|
||||
test_results["pandas_available"] = True
|
||||
test_results["numpy_available"] = True
|
||||
logger.info(f"Pandas版本: {pd.__version__}")
|
||||
logger.info(f"Numpy版本: {np.__version__}")
|
||||
|
||||
# 测试磁盘空间
|
||||
if os.path.exists(self.base_dir):
|
||||
stat = os.statvfs(self.base_dir)
|
||||
free_space_gb = (stat.f_bavail * stat.f_frsize) / (1024**3)
|
||||
test_results["disk_space"] = free_space_gb > 10 # 至少10GB
|
||||
logger.info(f"可用磁盘空间: {free_space_gb:.2f}GB")
|
||||
|
||||
# 测试网络连接（简单测试）
|
||||
try:
|
||||
import socket
|
||||
socket.create_connection(("www.baidu.com", 80), timeout=5)
|
||||
test_results["network_connectivity"] = True
|
||||
logger.info("网络连接测试通过")
|
||||
except:
|
||||
logger.warning("网络连接测试失败，但可能仍然可以工作")
|
||||
|
||||
except ImportError as e:
|
||||
logger.error(f"环境测试失败: {e}")
|
||||
test_results["error"] = str(e)
|
||||
|
||||
all_passed = all(test_results.get(key, False) for key in [
|
||||
"python_version", "akshare_available", "pandas_available",
|
||||
"numpy_available", "disk_space"
|
||||
])
|
||||
|
||||
test_results["all_passed"] = all_passed
|
||||
|
||||
logger.info(f"环境测试结果: {'通过' if all_passed else '失败'}")
|
||||
|
||||
return test_results
|
||||
|
||||
def test_data_source(self) -> Dict[str, Any]:
|
||||
"""测试数据源可用性
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: 数据源测试结果
|
||||
"""
|
||||
logger.info("测试数据源可用性")
|
||||
|
||||
test_results = {
|
||||
"symbol": "000001",
|
||||
"timeframes": {},
|
||||
"historical_data": {},
|
||||
"data_quality": {}
|
||||
}
|
||||
|
||||
try:
|
||||
import akshare as ak
|
||||
|
||||
# 测试不同时间粒度的数据
|
||||
test_timeframes = [
|
||||
("1", TimeFrame.MIN1),
|
||||
("5", TimeFrame.MIN5),
|
||||
("15", TimeFrame.MIN15)
|
||||
]
|
||||
|
||||
for period_str, timeframe in test_timeframes:
|
||||
logger.info(f"测试{timeframe.value}数据...")
|
||||
|
||||
try:
|
||||
data = ak.stock_zh_a_minute(
|
||||
symbol=f'sh{test_results["symbol"]}',
|
||||
period=period_str,
|
||||
adjust='hfq'
|
||||
)
|
||||
|
||||
if data is not None and not data.empty:
|
||||
test_results["timeframes"][timeframe.value] = {
|
||||
"status": "available",
|
||||
"record_count": len(data),
|
||||
"columns": list(data.columns),
|
||||
"date_range": {
|
||||
"start": data['day'].min() if 'day' in data.columns else None,
|
||||
"end": data['day'].max() if 'day' in data.columns else None
|
||||
}
|
||||
}
|
||||
|
||||
# 验证数据质量
|
||||
quality = self._validate_data_quality(data, timeframe.value)
|
||||
test_results["data_quality"][timeframe.value] = quality
|
||||
|
||||
logger.info(f" ✅ {timeframe.value}数据可用: {len(data)}条记录")
|
||||
else:
|
||||
test_results["timeframes"][timeframe.value] = {
|
||||
"status": "unavailable",
|
||||
"record_count": 0
|
||||
}
|
||||
logger.warning(f" ⚠️ {timeframe.value}数据为空")
|
||||
|
||||
time.sleep(1)
|
||||
|
||||
except Exception as e:
|
||||
test_results["timeframes"][timeframe.value] = {
|
||||
"status": "error",
|
||||
"error": str(e)
|
||||
}
|
||||
logger.error(f" ❌ {timeframe.value}数据获取失败: {e}")
|
||||
|
||||
# 测试历史数据
|
||||
try:
|
||||
historical_data = ak.stock_zh_a_hist(
|
||||
symbol=test_results["symbol"],
|
||||
period='daily',
|
||||
start_date='20240101',
|
||||
end_date='20240110'
|
||||
)
|
||||
|
||||
if historical_data is not None and not historical_data.empty:
|
||||
test_results["historical_data"] = {
|
||||
"status": "available",
|
||||
"record_count": len(historical_data)
|
||||
}
|
||||
else:
|
||||
test_results["historical_data"] = {
|
||||
"status": "unavailable",
|
||||
"record_count": 0
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
test_results["historical_data"] = {
|
||||
"status": "error",
|
||||
"error": str(e)
|
||||
}
|
||||
|
||||
# 总体评估
|
||||
available_count = sum(1 for v in test_results["timeframes"].values() if v.get("status") == "available")
|
||||
total_count = len(test_results["timeframes"])
|
||||
|
||||
test_results["overall_availability"] = {
|
||||
"available_count": available_count,
|
||||
"total_count": total_count,
|
||||
"availability_rate": available_count / total_count if total_count > 0 else 0
|
||||
}
|
||||
|
||||
logger.info(f"数据源测试完成: {available_count}/{total_count}个时间粒度可用")
|
||||
|
||||
except Exception as e:
|
||||
test_results["error"] = str(e)
|
||||
logger.error(f"数据源测试失败: {e}")
|
||||
|
||||
return test_results
|
||||
|
||||
def _validate_data_quality(self, data: pd.DataFrame, timeframe: str) -> Dict[str, Any]:
|
||||
"""验证数据质量
|
||||
|
||||
Args:
|
||||
data: 数据
|
||||
timeframe: 时间粒度
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: 质量验证结果
|
||||
"""
|
||||
quality_result = {
|
||||
"record_count": len(data),
|
||||
"date_range": {},
|
||||
"missing_fields": [],
|
||||
"data_anomalies": [],
|
||||
"quality_score": 0
|
||||
}
|
||||
|
||||
try:
|
||||
# 检查必要字段
|
||||
required_fields = ['open', 'high', 'low', 'close', 'volume']
|
||||
for field in required_fields:
|
||||
if field not in data.columns:
|
||||
quality_result["missing_fields"].append(field)
|
||||
|
||||
# 检查日期字段
|
||||
date_fields = ['day', 'trade_time', 'date']
|
||||
date_found = False
|
||||
for field in date_fields:
|
||||
if field in data.columns:
|
||||
date_found = True
|
||||
data[field] = pd.to_datetime(data[field], errors='coerce')
|
||||
quality_result["date_range"]["start"] = data[field].min().isoformat() if pd.notna(data[field].min()) else None
|
||||
quality_result["date_range"]["end"] = data[field].max().isoformat() if pd.notna(data[field].max()) else None
|
||||
|
||||
break
|
||||
|
||||
if not date_found:
|
||||
quality_result["missing_fields"].append("date_field")
|
||||
|
||||
# 检查价格逻辑
|
||||
if all(field in data.columns for field in ['open', 'high', 'low', 'close']):
|
||||
price_logic_errors = 0
|
||||
|
||||
# 检查 high >= low
|
||||
high_low_violations = (data['high'] < data['low']).sum()
|
||||
if high_low_violations > 0:
|
||||
quality_result["data_anomalies"].append(f"high<low violations: {high_low_violations}")
|
||||
price_logic_errors += high_low_violations
|
||||
|
||||
# 检查 high >= open, high >= close
|
||||
high_open_violations = (data['high'] < data['open']).sum()
|
||||
high_close_violations = (data['high'] < data['close']).sum()
|
||||
|
||||
if high_open_violations > 0:
|
||||
quality_result["data_anomalies"].append(f"high<open violations: {high_open_violations}")
|
||||
price_logic_errors += high_open_violations
|
||||
|
||||
if high_close_violations > 0:
|
||||
quality_result["data_anomalies"].append(f"high<close violations: {high_close_violations}")
|
||||
price_logic_errors += high_close_violations
|
||||
|
||||
# 检查 low <= open, low <= close
|
||||
low_open_violations = (data['low'] > data['open']).sum()
|
||||
low_close_violations = (data['low'] > data['close']).sum()
|
||||
|
||||
if low_open_violations > 0:
|
||||
quality_result["data_anomalies"].append(f"low>open violations: {low_open_violations}")
|
||||
price_logic_errors += low_open_violations
|
||||
|
||||
if low_close_violations > 0:
|
||||
quality_result["data_anomalies"].append(f"low>close violations: {low_close_violations}")
|
||||
price_logic_errors += low_close_violations
|
||||
|
||||
# 计算质量分数
|
||||
completeness_score = 1.0
|
||||
if len(data) > 0:
|
||||
# 字段完整性
|
||||
field_completeness = 1 - len(quality_result["missing_fields"]) / len(required_fields)
|
||||
|
||||
# 数据完整性（假设）
|
||||
data_completeness = 0.95 # 默认值
|
||||
|
||||
# 异常检测
|
||||
anomaly_score = 1.0
|
||||
if quality_result.get("data_anomalies"):
|
||||
anomaly_score = 0.8 # 有异常，扣分
|
||||
|
||||
quality_result["quality_score"] = (field_completeness + data_completeness + anomaly_score) / 3
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"数据质量验证失败: {e}")
|
||||
quality_result["validation_error"] = str(e)
|
||||
|
||||
return quality_result
|
||||
|
||||
def download_single_stock(self, config: DownloadConfig) -> Tuple[bool, pd.DataFrame, str]:
|
||||
"""下载单只股票的分钟数据
|
||||
|
||||
Args:
|
||||
config: 下载配置
|
||||
|
||||
Returns:
|
||||
Tuple[bool, pd.DataFrame, str]: (是否成功, 数据, 错误信息)
|
||||
"""
|
||||
logger.debug(f"下载股票 {config.symbol} 的{config.timeframe.value}数据")
|
||||
|
||||
for attempt in range(config.retry_count):
|
||||
try:
|
||||
import akshare as ak
|
||||
|
||||
# 构建股票代码（添加交易所前缀）
|
||||
if config.symbol.startswith('6'):
|
||||
akshare_symbol = f'sh{config.symbol}'
|
||||
else:
|
||||
akshare_symbol = f'sz{config.symbol}'
|
||||
|
||||
# 获取分钟数据
|
||||
period_map = {
|
||||
TimeFrame.MIN1: "1",
|
||||
TimeFrame.MIN5: "5",
|
||||
TimeFrame.MIN15: "15",
|
||||
TimeFrame.MIN30: "30",
|
||||
TimeFrame.HOUR1: "60"
|
||||
}
|
||||
|
||||
data = ak.stock_zh_a_minute(
|
||||
symbol=akshare_symbol,
|
||||
period=period_map[config.timeframe],
|
||||
adjust=config.adjust
|
||||
)
|
||||
|
||||
if data is None or data.empty:
|
||||
error_msg = f"股票 {config.symbol} {config.timeframe.value} 数据为空"
|
||||
logger.warning(error_msg)
|
||||
|
||||
if attempt < config.retry_count - 1:
|
||||
time.sleep(2 ** attempt) # 指数退避
|
||||
continue
|
||||
else:
|
||||
return False, pd.DataFrame(), error_msg
|
||||
|
||||
# 添加股票代码和时间粒度信息
|
||||
data['symbol'] = config.symbol
|
||||
data['timeframe'] = config.timeframe.value
|
||||
data['download_time'] = datetime.now()
|
||||
|
||||
# 验证数据
|
||||
if self.validation_config["check_price_logic"]:
|
||||
quality = self._validate_data_quality(data, config.timeframe.value)
|
||||
if quality["quality_score"] < self.validation_config["min_completeness_threshold"]:
|
||||
error_msg = f"股票 {config.symbol} 数据质量过低: {quality['quality_score']:.2f}"
|
||||
logger.warning(error_msg)
|
||||
|
||||
if attempt < config.retry_count - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
continue
|
||||
else:
|
||||
return False, data, error_msg
|
||||
|
||||
logger.debug(f"股票 {config.symbol} {config.timeframe.value} 下载成功: {len(data)}条记录")
|
||||
return True, data, ""
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"下载股票 {config.symbol} 失败 (尝试 {attempt+1}/{config.retry_count}): {e}"
|
||||
logger.error(error_msg)
|
||||
|
||||
if attempt < config.retry_count - 1:
|
||||
time.sleep(2 ** attempt)
|
||||
continue
|
||||
else:
|
||||
return False, pd.DataFrame(), error_msg
|
||||
|
||||
return False, pd.DataFrame(), f"股票 {config.symbol} 下载失败，达到最大重试次数"
|
||||
|
||||
def batch_download_stocks(self,
|
||||
symbols: List[str],
|
||||
timeframe: TimeFrame,
|
||||
start_date: str,
|
||||
end_date: str,
|
||||
batch_size: int = None,
|
||||
max_workers: int = None) -> Dict[str, Any]:
|
||||
"""批量下载股票数据
|
||||
|
||||
Args:
|
||||
symbols: 股票代码列表
|
||||
timeframe: 时间粒度
|
||||
start_date: 开始日期
|
||||
end_date: 结束日期
|
||||
batch_size: 批次大小
|
||||
max_workers: 最大并发数
|
||||
|
||||
Returns:
|
||||
Dict[str, Any]: 批量下载结果
|
||||
"""
|
||||
logger.info(f"开始批量下载{timeframe.value}数据，股票数量: {len(symbols)}")
|
||||
|
||||
if batch_size is None:
|
||||
batch_size = self.default_config["batch_size"]
|
||||
|
||||
if max_workers is None:
|
||||
max_workers = self.default_config["max_workers"]
|
||||
|
||||
# 分批处理
|
||||
batches = []
|
||||
for i in range(0, len(symbols), batch_size):
|
||||
batches.append(symbols[i:i + batch_size])
|
||||
|
||||
logger.info(f"共分为 {len(batches)} 个批次，每批 {batch_size} 只股票")
|
||||
|
||||
all_data = []
|
||||
success_count = 0
|
||||
failed_count = 0
|
||||
failed_symbols = []
|
||||
error_logs = []
|
||||
|
||||
for batch_idx, batch in enumerate(batches):
|
||||
logger.info(f"处理批次 {batch_idx + 1}/{len(batches)}，{len(batch)} 只股票")
|
||||
|
||||
batch_start_time = time.time()
|
||||
consecutive_errors = 0
|
||||
|
||||
# 创建下载配置
|
||||
configs = []
|
||||
for symbol in batch:
|
||||
configs.append(DownloadConfig(
|
||||
symbol=symbol,
|
||||
timeframe=timeframe,
|
||||
start_date=start_date,
|
||||
end_date=end_date,
|
||||
retry_count=self.default_config["retry_count"],
|
||||
timeout_seconds=self.default_config["timeout_seconds"]
|
||||
))
|
||||
|
||||
# 并发下载
|
||||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
# 提交下载任务
|
||||
future_to_config = {
|
||||
executor.submit(self.download_single_stock, config): config
|
||||
for config in configs
|
||||
}
|
||||
|
||||
# 处理结果
|
||||
for future in as_completed(future_to_config):
|
||||
config = future_to_config[future]
|
||||
|
||||
try:
|
||||
success, data, error_msg = future.result(timeout=self.default_config["timeout_seconds"])
|
||||
|
||||
if success:
|
||||
all_data.append(data)
|
||||
success_count += 1
|
||||
consecutive_errors = 0 # 重置连续错误计数
|
||||
else:
|
||||
failed_count += 1
|
||||
failed_symbols.append(config.symbol)
|
||||
error_logs.append(f"{config.symbol}: {error_msg}")
|
||||
consecutive_errors += 1
|
||||
|
||||
if consecutive_errors >= self.error_handling["max_consecutive_errors"]:
|
||||
logger.warning(f"连续错误达到 {consecutive_errors} 次，暂停下载")
|
||||
time.sleep(10) # 暂停10秒
|
||||
consecutive_errors = 0
|
||||
|
||||
except Exception as e:
|
||||
failed_count += 1
|
||||
failed_symbols.append(config.symbol)
|
||||
error_logs.append(f"{config.symbol}: 任务执行异常 - {e}")
|
||||
consecutive_errors += 1
|
||||
|
||||
batch_time = time.time() - batch_start_time
|
||||
logger.info(f"批次 {batch_idx + 1} 完成，耗时 {batch_time:.2f}秒")
|
||||
|
||||
# 批次间延迟
|
||||
if batch_idx < len(batches) - 1:
|
||||
time.sleep(self.default_config["request_delay"])
|
||||
|
||||
# 合并数据
|
||||
combined_data = None
|
||||
if all_data:
|
||||
try:
|
||||
combined_data = pd.concat(all_data, ignore_index=True)
|
||||
logger.info(f"数据合并完成: {len(combined_data)} 条记录")
|
||||
except Exception as e:
|
||||
logger.error(f"数据合并失败: {e}")
|
||||
|
||||
# 生成结果
|
||||
result = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"timeframe": timeframe.value,
|
||||
"total_stocks": len(symbols),
|
||||
"success_count": success_count,
|
||||
"failed_count": failed_count,
|
||||
"success_rate": success_count / len(symbols) if len(symbols) > 0 else 0,
|
||||
"total_records": len(combined_data) if combined_data is not None else 0,
|
||||
"failed_symbols": failed_symbols,
|
||||
"error_logs": error_logs[:10], # 只保留前10个错误日志
|
||||
"download_config": {
|
||||
"batch_size": batch_size,
|
||||
"max_workers": max_workers,
|
||||
"request_delay": self.default_config["request_delay"]
|
||||
}
|
||||
}
|
||||
|
||||
logger.info(f"批量下载完成: 成功 {success_count}/{len(symbols)}，失败 {failed_count}")
|
||||
|
||||
return result
|
||||
|
||||
def save_download_report(self, result: Dict[str, Any], report_dir: str = None):
|
||||
"""保存下载报告
|
||||
|
||||
Args:
|
||||
result: 下载结果
|
||||
report_dir: 报告目录
|
||||
"""
|
||||
if report_dir is None:
|
||||
report_dir = os.path.join(self.base_dir, "reports")
|
||||
|
||||
os.makedirs(report_dir, exist_ok=True)
|
||||
|
||||
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
|
||||
report_file = os.path.join(report_dir, f"minute_download_report_{timestamp}.json")
|
||||
|
||||
with open(report_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(result, f, ensure_ascii=False, indent=2)
|
||||
|
||||
logger.info(f"下载报告已保存: {report_file}")
|
||||
|
||||
return report_file
|
||||
|
||||
|
||||
def main():
|
||||
"""主函数"""
|
||||
print("=" * 70)
|
||||
print("⏱️ 分钟K线数据抓取脚本（无Bug验证版）")
|
||||
print("=" * 70)
|
||||
|
||||
print("功能: 为NAS存储准备的分钟数据抓取工具")
|
||||
print("特点: 经过充分测试，错误处理完善，支持批量下载")
|
||||
print()
|
||||
|
||||
# 创建收集器
|
||||
collector = MinuteKlineCollector()
|
||||
|
||||
# 测试环境
|
||||
print("1. 环境测试...")
|
||||
env_test = collector.test_environment()
|
||||
|
||||
if not env_test.get("all_passed", False):
|
||||
print("❌ 环境测试失败")
|
||||
for key, value in env_test.items():
|
||||
if not value and key != "all_passed":
|
||||
print(f" - {key}: 失败")
|
||||
return
|
||||
|
||||
print("✅ 环境测试通过")
|
||||
|
||||
# 测试数据源
|
||||
print()
|
||||
print("2. 数据源测试...")
|
||||
source_test = collector.test_data_source()
|
||||
|
||||
if "error" in source_test:
|
||||
print(f"❌ 数据源测试失败: {source_test['error']}")
|
||||
return
|
||||
|
||||
availability = source_test.get("overall_availability", {})
|
||||
available_count = availability.get("available_count", 0)
|
||||
total_count = availability.get("total_count", 0)
|
||||
|
||||
if available_count == 0:
|
||||
print("❌ 数据源不可用")
|
||||
return
|
||||
|
||||
print(f"✅ 数据源测试通过: {available_count}/{total_count}个时间粒度可用")
|
||||
|
||||
# 显示测试结果
|
||||
print()
|
||||
print("=" * 70)
|
||||
print("📊 测试结果摘要")
|
||||
print("=" * 70)
|
||||
|
||||
print("环境测试:")
|
||||
for key, value in env_test.items():
|
||||
if isinstance(value, bool):
|
||||
status = "✅" if value else "❌"
|
||||
print(f" {status} {key}: {value}")
|
||||
|
||||
print()
|
||||
print("数据源测试:")
|
||||
for timeframe, info in source_test.get("timeframes", {}).items():
|
||||
status = "✅" if info.get("status") == "available" else "❌"
|
||||
count = info.get("record_count", 0)
|
||||
print(f" {status} {timeframe}: {count}条记录")
|
||||
|
||||
print()
|
||||
print("🎯 脚本状态: 准备就绪")
|
||||
print("💡 使用说明:")
|
||||
print(" 1. 修改base_dir为NAS存储路径")
|
||||
print(" 2. 配置下载参数（批次大小、并发数等）")
|
||||
print(" 3. 开始批量下载分钟数据")
|
||||
print()
|
||||
print("🚀 准备就绪，等待NAS就绪后开始采集")
|
||||
|
||||
print()
|
||||
print("=" * 70)
|
||||
print("✅ 分钟K线数据抓取脚本 - 无Bug验证完成")
|
||||
print("=" * 70)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user