19年創業做過一年的量化交易但沒有成功,作為交易系統的開發人員積累了一些經驗,最近想重新研究交易系統,一邊整理一邊寫出來一些思考供大家參考,也希望跟做量化的朋友有更多的交流和合作。
接下來會對于真格(澎博財經旗下)平臺介紹。
真格平臺是澎博財經旗下的一款量化交易工具,專注于提供高效的數據接口、量化策略開發環境和回測框架。通過真格平臺,用戶可以方便地構建、測試和優化量化交易策略,同時利用其專業化工具高效地管理交易流程。
以下以“布林帶策略”為例,展示如何使用真格平臺進行量化交易開發和回測。
1. 策略背景:布林帶策略
策略邏輯
布林帶是一種經典的技術指標,通過價格與波動率之間的關系生成交易信號。布林帶由以下三部分構成:
- 上軌線:均線 + K倍標準差
- 中軌線:簡單移動平均線
- 下軌線:均線 - K倍標準差
布林帶策略邏輯如下:
- 買入信號:價格觸及下軌并開始反彈。
- 賣出信號:價格觸及上軌并開始回落。
適用場景
- 優勢:適用于震蕩市場,能夠捕捉價格的區間波動。
- 局限:在強趨勢行情中,可能錯過趨勢性機會。
2. 策略開發
以下代碼基于真格平臺Python SDK實現布林帶策略的開發與回測。
(1)準備工作
在使用真格平臺之前,需要完成以下準備:
- 安裝SDK:通過真格平臺提供的安裝包或pip安裝其Python SDK。
pip install zhenge-sdk
- 申請開發者賬號:獲取API密鑰以訪問平臺數據接口。
- 設置交易環境:確認回測所需的賬戶、品種和時間段。
(2)初始化參數與環境
import pandas as pd
import numpy as np
from zhenge import ZhengeAPI, Backtest# 策略參數
SYMBOL = '600519.SH' # 茅臺股票
START_DATE = '2020-01-01'
END_DATE = '2022-01-01'
INITIAL_CAPITAL = 100000 # 初始資金
BOLL_PERIOD = 20 # 布林帶周期
BOLL_K = 2 # 布林帶倍數
(3)獲取歷史數據
通過真格平臺的API獲取標的歷史行情數據。
# 獲取歷史數據
def get_data(symbol, start_date, end_date):api = ZhengeAPI(api_key='your_api_key')df = api.get_kline(symbol=symbol, start_date=start_date, end_date=end_date, freq='1d')df['Date'] = pd.to_datetime(df['date'])df.set_index('Date', inplace=True)return df[['close']]# 加載數據
data = get_data(SYMBOL, START_DATE, END_DATE)
(4)計算布林帶
利用Pandas計算布林帶的上軌、中軌和下軌。
# 計算布林帶
data['Middle'] = data['close'].rolling(BOLL_PERIOD).mean()
data['StdDev'] = data['close'].rolling(BOLL_PERIOD).std()
data['Upper'] = data['Middle'] + BOLL_K * data['StdDev']
data['Lower'] = data['Middle'] - BOLL_K * data['StdDev']
(5)生成交易信號
根據布林帶策略邏輯生成交易信號。
# 生成交易信號
data['Signal'] = 0
data.loc[data['close'] < data['Lower'], 'Signal'] = 1 # 買入信號
data.loc[data['close'] > data['Upper'], 'Signal'] = -1 # 賣出信號
(6)回測邏輯
通過信號生成持倉和資金變化。
# 初始化回測變量
data['Position'] = 0 # 持倉
data['Cash'] = INITIAL_CAPITAL # 初始現金
data['Portfolio'] = INITIAL_CAPITAL # 總資產holdings = 0 # 當前持倉數量
entry_price = 0 # 買入價格# 回測主循環
for i in range(1, len(data)):current_signal = data.iloc[i]['Signal']current_price = data.iloc[i]['close']# 買入邏輯if current_signal == 1 and holdings == 0:holdings = data.iloc[i - 1]['Cash'] // current_pricedata.at[data.index[i], 'Cash'] = data.iloc[i - 1]['Cash'] - holdings * current_pricedata.at[data.index[i], 'Position'] = holdingsentry_price = current_price # 記錄買入價格# 賣出邏輯elif current_signal == -1 and holdings > 0:data.at[data.index[i], 'Cash'] = data.iloc[i - 1]['Cash'] + holdings * current_priceholdings = 0data.at[data.index[i], 'Position'] = holdings# 更新總資產current_holdings = holdings * current_pricedata.at[data.index[i], 'Portfolio'] = data.iloc[i]['Cash'] + current_holdings
3. 策略優化
(1)參數優化
通過調整布林帶周期和倍數,優化策略參數。
best_params = None
best_return = -float('inf')for period in range(10, 31):for k in np.arange(1.5, 3.0, 0.1):# 計算布林帶data['Middle'] = data['close'].rolling(period).mean()data['StdDev'] = data['close'].rolling(period).std()data['Upper'] = data['Middle'] + k * data['StdDev']data['Lower'] = data['Middle'] - k * data['StdDev']# 生成信號data['Signal'] = 0data.loc[data['close'] < data['Lower'], 'Signal'] = 1data.loc[data['close'] > data['Upper'], 'Signal'] = -1# 簡化回測邏輯holdings = 0cash = INITIAL_CAPITALfor i in range(1, len(data)):signal = data.iloc[i]['Signal']price = data.iloc[i]['close']if signal == 1 and holdings == 0:holdings = cash // pricecash -= holdings * priceelif signal == -1 and holdings > 0:cash += holdings * priceholdings = 0portfolio = cash + holdings * data.iloc[-1]['close']total_return = (portfolio - INITIAL_CAPITAL) / INITIAL_CAPITALif total_return > best_return:best_return = total_returnbest_params = (period, k)print(f"最佳參數:周期={best_params[0]}, 倍數={best_params[1]:.2f}, 收益率={best_return:.2%}")
(2)加入風險管理
在策略中添加止盈和止損機制。
TAKE_PROFIT = 0.1 # 止盈10%
STOP_LOSS = 0.05 # 止損5%for i in range(1, len(data)):if holdings > 0:pnl = (data.iloc[i]['close'] - entry_price) / entry_priceif pnl >= TAKE_PROFIT or pnl <= -STOP_LOSS:data.at[data.index[i], 'Cash'] = data.iloc[i - 1]['Cash'] + holdings * data.iloc[i]['close']holdings = 0data.at[data.index[i], 'Position'] = holdings