# 交易策略建構範例
提示
- 以下交易策略不構成投資建議，僅供學習參考。
# 策略概述
構建一個雙均線策略：
運用某一標的1分 K 線，計算出兩條不同週期的移動平均線 MA1 和 MA3，跟蹤 MA1 和 MA3 的相對大小，由此判斷買賣時機。
當 MA1 >= MA3 時，判斷該標的為強勢狀態，市場屬於多頭市場，採取開倉的操作；
當 MA1 < MA3 時，判斷該標的為弱勢狀態，市場屬於空頭市場，採取平倉的操作。
# 流程圖
# 程式碼範例
- Example
from futu import *
############################ 全域變數設定 ############################
FUTUOPEND_ADDRESS = '127.0.0.1' # OpenD 監聽地址
FUTUOPEND_PORT = 11111 # OpenD 監聽連接埠
TRADING_ENVIRONMENT = TrdEnv.SIMULATE # 交易環境：真實 / 模擬
TRADING_MARKET = TrdMarket.HK # 交易市場權限，用於篩選對應交易市場權限的帳戶
TRADING_PWD = '123456' # 交易密碼，用於解鎖交易
TRADING_PERIOD = KLType.K_1M # 信號 K 線週期
TRADING_SECURITY = 'HK.00700' # 交易標的
FAST_MOVING_AVERAGE = 1 # 均線快線的週期
SLOW_MOVING_AVERAGE = 3 # 均線慢線的週期
quote_context = OpenQuoteContext(host=FUTUOPEND_ADDRESS, port=FUTUOPEND_PORT) # 行情物件
trade_context = OpenSecTradeContext(filter_trdmarket=TRADING_MARKET, host=FUTUOPEND_ADDRESS, port=FUTUOPEND_PORT, security_firm=SecurityFirm.FUTUSECURITIES) # 交易物件，根據交易產品修改交易物件類型
# 解鎖交易
def unlock_trade():
if TRADING_ENVIRONMENT == TrdEnv.REAL:
ret, data = trade_context.unlock_trade(TRADING_PWD)
if ret != RET_OK:
print('解鎖交易失敗：', data)
return False
print('解鎖交易成功！')
return True
# 獲取市場狀態
def is_normal_trading_time(code):
ret, data = quote_context.get_market_state([code])
if ret != RET_OK:
print('獲取市場狀態失敗：', data)
return False
market_state = data['market_state'][0]
'''
MarketState.MORNING 港、A 股早盤
MarketState.AFTERNOON 港、A 股下午盤，美股全天
MarketState.FUTURE_DAY_OPEN 港、新、日期貨日市開盤
MarketState.FUTURE_OPEN 美期貨開盤
MarketState.FUTURE_BREAK_OVER 美期貨休息後開盤
MarketState.NIGHT_OPEN 港、新、日期貨夜市開盤
'''
if market_state == MarketState.MORNING or \
market_state == MarketState.AFTERNOON or \
market_state == MarketState.FUTURE_DAY_OPEN or \
market_state == MarketState.FUTURE_OPEN or \
market_state == MarketState.FUTURE_BREAK_OVER or \
market_state == MarketState.NIGHT_OPEN:
return True
print('現在不是持續交易時段。')
return False
# 獲取持倉數量
def get_holding_position(code):
holding_position = 0
ret, data = trade_context.position_list_query(code=code, trd_env=TRADING_ENVIRONMENT)
if ret != RET_OK:
print('獲取持倉數據失敗：', data)
return None
else:
for qty in data['qty'].values.tolist():
holding_position += qty
print('【持倉狀態】 {} 的持倉數量為：{}'.format(TRADING_SECURITY, holding_position))
return holding_position
# 拉取 K 線，計算均線，判斷多空
def calculate_bull_bear(code, fast_param, slow_param):
if fast_param <= 0 or slow_param <= 0:
return 0
if fast_param > slow_param:
return calculate_bull_bear(code, slow_param, fast_param)
ret, data = quote_context.get_cur_kline(code=code, num=slow_param + 1, ktype=TRADING_PERIOD)
if ret != RET_OK:
print('獲取K線失敗：', data)
return 0
candlestick_list = data['close'].values.tolist()[::-1]
fast_value = None
slow_value = None
if len(candlestick_list) > fast_param:
fast_value = sum(candlestick_list[1: fast_param + 1]) / fast_param
if len(candlestick_list) > slow_param:
slow_value = sum(candlestick_list[1: slow_param + 1]) / slow_param
if fast_value is None or slow_value is None:
return 0
return 1 if fast_value >= slow_value else -1
# 獲取一檔擺盤的 ask1 和 bid1
def get_ask_and_bid(code):
ret, data = quote_context.get_order_book(code, num=1)
if ret != RET_OK:
print('獲取擺盤數據失敗：', data)
return None, None
return data['Ask'][0][0], data['Bid'][0][0]
# 開倉函數
def open_position(code):
# 獲取擺盤數據
ask, bid = get_ask_and_bid(code)
# 計算下單量
open_quantity = calculate_quantity()
# 判斷購買力是否足夠
if is_valid_quantity(TRADING_SECURITY, open_quantity, ask):
# 下單
ret, data = trade_context.place_order(price=ask, qty=open_quantity, code=code, trd_side=TrdSide.BUY,
order_type=OrderType.NORMAL, trd_env=TRADING_ENVIRONMENT,
remark='moving_average_strategy')
if ret != RET_OK:
print('開倉失敗：', data)
else:
print('下單數量超出最大可買數量。')
# 平倉函數
def close_position(code, quantity):
# 獲取擺盤數據
ask, bid = get_ask_and_bid(code)
# 檢查平倉數量
if quantity == 0:
print('無效的下單數量。')
return False
# 平倉
ret, data = trade_context.place_order(price=bid, qty=quantity, code=code, trd_side=TrdSide.SELL,
order_type=OrderType.NORMAL, trd_env=TRADING_ENVIRONMENT, remark='moving_average_strategy')
if ret != RET_OK:
print('平倉失敗：', data)
return False
return True
# 計算下單數量
def calculate_quantity():
price_quantity = 0
# 使用最小交易量
ret, data = quote_context.get_market_snapshot([TRADING_SECURITY])
if ret != RET_OK:
print('獲取快照失敗：', data)
return price_quantity
price_quantity = data['lot_size'][0]
return price_quantity
# 判斷購買力是否足夠
def is_valid_quantity(code, quantity, price):
ret, data = trade_context.acctradinginfo_query(order_type=OrderType.NORMAL, code=code, price=price,
trd_env=TRADING_ENVIRONMENT)
if ret != RET_OK:
print('獲取最大可買可賣失敗：', data)
return False
max_can_buy = data['max_cash_buy'][0]
max_can_sell = data['max_sell_short'][0]
if quantity > 0:
return quantity < max_can_buy
elif quantity < 0:
return abs(quantity) < max_can_sell
else:
return False
# 展示訂單回呼
def show_order_status(data):
order_status = data['order_status'][0]
order_info = dict()
order_info['代碼'] = data['code'][0]
order_info['價格'] = data['price'][0]
order_info['方向'] = data['trd_side'][0]
order_info['數量'] = data['qty'][0]
print('【訂單狀態】', order_status, order_info)
############################ 填充以下函數來完成您的策略 ############################
# 策略啓動時執行一次，用於初始化策略
def on_init():
# 解鎖交易（如果是模擬交易則不需要解鎖）
if not unlock_trade():
return False
print('************ 策略開始執行 ***********')
return True
# 每個 tick 執行一次，可將策略的主要邏輯寫在此處
def on_tick():
pass
# 每次產生一根新的 K 線執行一次，可將策略的主要邏輯寫在此處
def on_bar_open():
# 列印分隔線
print('*************************************')
# 只在常規交易時段交易
if not is_normal_trading_time(TRADING_SECURITY):
return
# 獲取 K 線，計算均線，判斷多空
bull_or_bear = calculate_bull_bear(TRADING_SECURITY, FAST_MOVING_AVERAGE, SLOW_MOVING_AVERAGE)
# 獲取持倉數量
holding_position = get_holding_position(TRADING_SECURITY)
# 下單判斷
if holding_position == 0:
if bull_or_bear == 1:
print('【操作信號】 做多信號，建立多單。')
open_position(TRADING_SECURITY)
else:
print('【操作信號】 做空信號，不開空單。')
elif holding_position > 0:
if bull_or_bear == -1:
print('【操作信號】 做空信號，平掉持倉。')
close_position(TRADING_SECURITY, holding_position)
else:
print('【操作信號】 做多信號，無需加倉。')
# 委託成交有變化時執行一次
def on_fill(data):
pass
# 訂單狀態有變化時執行一次
def on_order_status(data):
if data['code'][0] == TRADING_SECURITY:
show_order_status(data)
################################ 框架實現部分，可忽略不看 ###############################
class OnTickClass(TickerHandlerBase):
def on_recv_rsp(self, rsp_pb):
on_tick()
class OnBarClass(CurKlineHandlerBase):
last_time = None
def on_recv_rsp(self, rsp_pb):
ret_code, data = super(OnBarClass, self).on_recv_rsp(rsp_pb)
if ret_code == RET_OK:
cur_time = data['time_key'][0]
if cur_time != self.last_time and data['k_type'][0] == TRADING_PERIOD:
if self.last_time is not None:
on_bar_open()
self.last_time = cur_time
class OnOrderClass(TradeOrderHandlerBase):
def on_recv_rsp(self, rsp_pb):
ret, data = super(OnOrderClass, self).on_recv_rsp(rsp_pb)
if ret == RET_OK:
on_order_status( data)
class OnFillClass(TradeDealHandlerBase):
def on_recv_rsp(self, rsp_pb):
ret, data = super(OnFillClass, self).on_recv_rsp(rsp_pb)
if ret == RET_OK:
on_fill(data)
# 主函式
if __name__ == '__main__':
# 初始化策略
if not on_init():
print('策略初始化失敗，腳本退出！')
quote_context.close()
trade_context.close()
else:
# 設定回呼
quote_context.set_handler(OnTickClass())
quote_context.set_handler(OnBarClass())
trade_context.set_handler(OnOrderClass())
trade_context.set_handler(OnFillClass())
# 訂閲標的合約的 逐筆，K 線和擺盤，以便獲取數據
quote_context.subscribe(code_list=[TRADING_SECURITY], subtype_list=[SubType.TICKER, SubType.ORDER_BOOK, TRADING_PERIOD])
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
- Output
************ 策略開始執行 ***********
*************************************
【持倉狀態】 HK.00700 的持倉數量為：0
【操作信號】 做多信號，建立多單。
【訂單狀態】 SUBMITTING {'代碼': 'HK.00700', '價格': 597.5, '方向': 'BUY', '數量': 100.0}
【訂單狀態】 SUBMITTED {'代碼': 'HK.00700', '價格': 597.5, '方向': 'BUY', '數量': 100.0}
【訂單狀態】 FILLED_ALL {'代碼': 'HK.00700', '價格': 597.5, '方向': 'BUY', '數量': 100.0}
*************************************
【持倉狀態】 HK.00700 的持倉數量為：100.0
【操作信號】 做空信號，平掉持倉。
【訂單狀態】 SUBMITTING {'代碼': 'HK.00700', '價格': 596.5, '方向': 'SELL', '數量': 100.0}
【訂單狀態】 SUBMITTED {'代碼': 'HK.00700', '價格': 596.5, '方向': 'SELL', '數量': 100.0}
【訂單狀態】 FILLED_ALL {'代碼': 'HK.00700', '價格': 596.5, '方向': 'SELL', '數量': 100.0}
1
2
3
4
5
6
7
8
9
10
11
12
13
2
3
4
5
6
7
8
9
10
11
12
13