#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
获取SHFE.rb2605的5分钟K线数据示例
"""
import time
import pandas as pd
from tqsdk import TqApi, TqAuth
def get_5min_kline():
"""
获取SHFE.rb2605的5分钟K线数据
"""
print("开始获取SHFE.rb2605的5分钟K线数据...")
try:
# 创建TqApi实例,使用指定的账号和密码
print("正在连接TqSdk...")
api = TqApi(auth=TqAuth("", ""))
print("连接TqSdk成功!")
# 定义合约代码
contract = "SHFE.rb2605"
# 定义K线周期(5分钟 = 300秒)
duration_seconds = 5 * 60
# 定义获取的K线数量
data_length = 100
print(f"正在获取 {contract} 的5分钟K线数据...")
print(f"K线周期: 5分钟,获取数量: {data_length}根")
# 获取K线数据
kline = api.get_kline_serial(contract, duration_seconds, data_length=data_length)
print("成功创建K线对象")
# 等待数据加载
print("等待数据加载...")
start_time = time.time()
if api.wait_update(deadline=start_time + 60):
print(f"数据加载完成,耗时: {time.time() - start_time:.2f}秒")
print(f"成功获取 {len(kline)} 条K线数据")
# 转换为DataFrame
df = pd.DataFrame({
"datetime": kline["datetime"],
"open": kline["open"],
"high": kline["high"],
"low": kline["low"],
"close": kline["close"],
"volume": kline["volume"]
})
# 转换时间戳为可读格式
df["datetime"] = pd.to_datetime(df["datetime"], unit="ns")
# 打印前5条数据
print("\n前5条K线数据:")
print(df.head())
# 打印数据统计信息
print("\n数据统计信息:")
print(f"数据时间范围: {df['datetime'].min()} 至 {df['datetime'].max()}")
print(f"收盘价范围: {df['close'].min():.2f} 至 {df['close'].max():.2f}")
print(f"平均成交量: {df['volume'].mean():.0f}")
# 保存数据到CSV文件
csv_filename = f"{contract}_5min_kline.csv"
df.to_csv(csv_filename, index=False)
print(f"\n数据已保存到 {csv_filename}")
return df
else:
print("数据加载超时")
return pd.DataFrame()
except Exception as e:
print(f"获取K线数据失败: {e}")
import traceback
traceback.print_exc()
return pd.DataFrame()
finally:
# 关闭API
try:
api.close()
print("\nAPI关闭成功!")
except:
pass
def main():
"""
主函数
"""
# 获取5分钟K线数据
df = get_5min_kline()
if not df.empty:
print("\n获取K线数据成功!")
else:
print("\n获取K线数据失败,请检查网络连接或合约代码是否正确。")
if __name__ == "__main__":
main()
有其他问题欢迎加入官方Q群748265037一起交流