量化交易系统开发-实时行情自动化交易-3.4.3.4.期货衍生数据
19年创业做过一年的量化交易但没有成功，作为交易系统的开发人员积累了一些经验，最近想重新研究交易系统，一边整理一边写出来一些思考供大家参考，也希望跟做量化的朋友有更多的交流和合作。
接下来聊聊基于期货API获取衍生数据。
衍生数据（Derived Data）是通过对原始市场数据进行计算和处理得出的数据，例如技术指标、资金流向、波动率等。这些数据有助于深入了解市场的动态、预测价格走向，并构建和优化交易策略。以下是通过 Python 编写的代码示例，利用期货公开 API（如和讯网、上海期货交易所等）获取并计算期货市场的衍生数据的详细开发内容。
1. 获取原始数据并计算技术指标
技术指标是常用的衍生数据之一，例如移动平均线（MA）、相对强弱指数（RSI）、布林带（Bollinger Bands）等。以下示例展示了如何通过和讯 API 获取 K 线数据并计算移动平均线和 RSI。
import requests
import pandas as pd
import numpy as np
def get_hexun_futures_klines(futures_code, limit=100):
"""
获取和讯期货的 K 线数据。
:param futures_code: 期货代码，例如 'AU0' 表示沪金连续合约
:param limit: 获取的 K 线数据数量
:return: K 线数据的 pandas DataFrame
"""
url = f"https://api.hexun.com/futures/kline"
params = {
"code": futures_code,
"type": "day", # 'day' 表示日 K 线，可以修改为 'week' 或 'month'
"count": limit
}
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
kline_data = data.get("data", {}).get("klines", [])
kline_list = [kline.split(",") for kline in kline_data]
df = pd.DataFrame(kline_list, columns=["日期", "开盘价", "最高价", "最低价", "收盘价", "成交量"])
df = df.astype({"开盘价": 'float', "最高价": 'float', "最低价": 'float', "收盘价": 'float', "成交量": 'int'})
return df
else:
raise Exception(f"Error fetching K line data: {response.status_code}")
# 获取沪金连续合约（AU0）的 K 线数据
df_klines = get_hexun_futures_klines("AU0")
# 计算移动平均线（MA）
df_klines["MA_5"] = df_klines["收盘价"].rolling(window=5).mean()
df_klines["MA_10"] = df_klines["收盘价"].rolling(window=10).mean()
# 计算相对强弱指数（RSI）
def calculate_rsi(data, window=14):
delta = data.diff(1)
gain = np.where(delta > 0, delta, 0)
loss = np.where(delta < 0, -delta, 0)
avg_gain = pd.Series(gain).rolling(window=window).mean()
avg_loss = pd.Series(loss).rolling(window=window).mean()
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
return rsi
df_klines["RSI_14"] = calculate_rsi(df_klines["收盘价"])
print(df_klines)在该示例中，我们通过调用和讯的 API 获取了指定期货合约的 K 线数据，并计算了 5 日和 10 日的移动平均线（MA）以及 14 日的相对强弱指数（RSI）。这些技术指标可以帮助识别市场的超买或超卖状态以及价格趋势。
2. 资金流向的计算
资金流向是衡量市场中资金流入和流出的重要指标，通过分析大单买入和卖出来评估市场中的资金动向。以下是通过和讯 API 获取逐笔交易数据并计算资金流向的示例。
def get_hexun_futures_trades(futures_code, limit=100):
"""
获取和讯期货的逐笔交易数据。
:param futures_code: 期货代码，例如 'AU0'
:param limit: 获取交易数据的数量
:return: 交易数据的 pandas DataFrame
"""
url = f"https://api.hexun.com/futures/trades"
params = {
"code": futures_code,
"count": limit
}
response = requests.get(url, params=params)
if response.status_code == 200:
data = response.json()
trades_data = data.get("data", {}).get("trades", [])
trades_list = [trade.split(",") for trade in trades_data]
df = pd.DataFrame(trades_list, columns=["时间", "价格", "数量", "买卖方向"])
df = df.astype({"价格": 'float', "数量": 'int'})
return df
else:
raise Exception(f"Error fetching trade data: {response.status_code}")
# 获取沪金连续合约（AU0）的逐笔交易数据
df_trades = get_hexun_futures_trades("AU0")
# 计算资金流向
df_trades["资金流向"] = df_trades.apply(lambda row: row["价格"] * row["数量"] if row["买卖方向"] == '买' else -row["价格"] * row["数量"], axis=1)
print(df_trades)在该示例中，通过和讯的 API 获取了期货的逐笔交易数据，并通过买卖方向计算出资金流向。资金流向为正表示资金流入，负值表示资金流出。
3. 波动率的计算
波动率是衡量市场价格波动剧烈程度的重要指标。以下代码展示了如何通过获取期货的 K 线数据计算其历史波动率。
# 计算历史波动率
def calculate_historical_volatility(data, window=20):
"""
计算期货的历史波动率。
:param data: 期货收盘价数据
:param window: 波动率计算的窗口期
:return: 历史波动率
"""
log_returns = np.log(data / data.shift(1))
volatility = log_returns.rolling(window=window).std() * np.sqrt(252) # 年化波动率
return volatility
# 计算沪金连续合约的 20 日历史波动率
df_klines["Volatility_20"] = calculate_historical_volatility(df_klines["收盘价"])
print(df_klines[["日期", "收盘价", "Volatility_20"]])在此示例中，通过计算期货收盘价的对数收益率并在滚动窗口上计算标准差，得到期货的历史波动率。这可以帮助交易者理解价格波动的剧烈程度，评估潜在风险。
4. 数据存储与管理
-
内存缓存：对于实时性的衍生数据，可以使用 Redis 等内存数据库进行缓存，以便于快速访问。
-
持久化存储：对于历史技术指标、资金流向等衍生数据，可以将其存储到 MySQL 或 InfluxDB 中，以方便后续分析和策略回测。
import mysql.connector def save_derived_data_to_mysql(df, futures_code, table_name): """ 将衍生数据保存到 MySQL 数据库中。 :param df: 衍生数据 DataFrame :param futures_code: 期货代码 :param table_name: 数据表名称 """ connection = mysql.connector.connect( host="localhost", user="root", password="password", database="futures_data" ) cursor = connection.cursor() create_table_query = f""" CREATE TABLE IF NOT EXISTS {table_name}_{futures_code} ( 日期 VARCHAR(20), 收盘价 FLOAT, MA_5 FLOAT, MA_10 FLOAT, RSI_14 FLOAT, Volatility_20 FLOAT )""" cursor.execute(create_table_query) for _, row in df.iterrows(): insert_query = f""" INSERT INTO {table_name}_{futures_code} (日期, 收盘价, MA_5, MA_10, RSI_14, Volatility_20) VALUES ('{row['日期']}', {row['收盘价']}, {row['MA_5']}, {row['MA_10']}, {row['RSI_14']}, {row['Volatility_20']}) """ cursor.execute(insert_query) connection.commit() cursor.close() connection.close() # 将衍生数据保存到 MySQL 数据库 save_derived_data_to_mysql(df_klines, "AU0", "derived_data")