19年创业做过一年的量化交易但没有成功，作为交易系统的开发人员积累了一些经验，最近想重新研究交易系统，一边整理一边写出来一些思考供大家参考，也希望跟做量化的朋友有更多的交流和合作。

接下来聊聊基于期货API获取衍生数据。

衍生数据（Derived Data）是通过对原始市场数据进行计算和处理得出的数据，例如技术指标、资金流向、波动率等。这些数据有助于深入了解市场的动态、预测价格走向，并构建和优化交易策略。以下是通过 Python 编写的代码示例，利用期货公开 API（如和讯网、上海期货交易所等）获取并计算期货市场的衍生数据的详细开发内容。

1. 获取原始数据并计算技术指标

技术指标是常用的衍生数据之一，例如移动平均线（MA）、相对强弱指数（RSI）、布林带（Bollinger Bands）等。以下示例展示了如何通过和讯 API 获取 K 线数据并计算移动平均线和 RSI。

import requests
import pandas as pd
import numpy as npdef 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 dfelse: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_lossrsi = 100 - (100 / (1 + rs))return rsidf_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 dfelse: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.connectordef 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")