本篇文章所使用的資料，來源于JQData本地量化金融資料庫。下面我將粗略的介紹一個強化學習在證券市場中應用的簡單實體。
關于強化學習的演算法理論及發展歷史，我們不做過多的解釋。我們可以很容易在互聯網上找到強化學習的理論知識，雖然可能都是一些只言片語，但對于初學者來說基本也就夠用了。到目前為止，還沒有出現廣受業內好評的中文教材，更多的參考資料還是英文版的。例如，Richard S.Sutton和Andrew G.Barto所著的《Reinforcement Learning: An Introduction》。這是比較好的強化學習教材，想要系統的、深入的學習強化學習，這本書值得一看。雖然國內學術界有很多關于強化學習的文章，但它們都看起來比較專業，我不建議初學者一上來就開始啃理論。最好的學習方式是你先入門，弄懂強化學習可以干什么？然后應用一些簡單的演算法搭建一個你當前正想解決的問題，再不斷的去改進你的演算法，并在這個程序中深入地學習。對于這篇文章而言，我們假設你已經有了一些強化學習的基礎知識了，這里只是給出了一個十分簡單的關于量化分析的應用demo而已。
作為量化分析領域的專業人員，我們可能對用強化學習解決玩游戲、找寶藏的Demo不感興趣。我們更希望能夠有一個簡單的強化學習demo：當輸入K線資料，就可以告訴我什么時候該買，什么時候該賣，即使給出的買賣點并不準確，但我們總算可以看看強化學習模型是怎么給出這個買賣點的。這篇文章就做了這樣一個demo，主要是想介紹怎樣在構建證券市場構建一個簡單的強化學習模型。
強化學習相比于神經網路等常見的機器學習演算法而言，強化學習更靈活多變。深度神經網路、卷積神經網路已經算是比較難的演算法了，但對于應用人員來說，你只需要搞懂輸入輸出基本就能用了。但強化學習完全不行，必須要對特征的問題抽象建模，這往往是最難的。怎樣從一堆證券資料中抽象各種各樣的狀態，以及這些狀態是怎么轉換的，怎么定義動作、回報等等。這些問題直接決定你的模型的質量。
在這篇文章中，我們需要解決的問題是：怎么利用一天內的48根5分鐘的K線資料探索在每個5分鐘結束的時候，我們是該買入（B），還是該賣出（S），或者是繼續觀望（W），并用一段時間內所有的5分鐘資料訓練這個模型，看哪個時間點最適合買入，哪個時間點最適合賣出。我們以時間點作為一個狀態標識，則狀態(S)轉移就比較好定義了：935（早上9點35，這個時間點產生了第一根K線）->940->945->…->1455->1500，狀態s->s’可以采取的動作(A)包含B、S、W。我們使用Q-learning演算法來解決這個問題。因此，Q表應該是這樣的：
關于Reward，我們是這樣定義的：未來一段時間的收益率，比如未來3根K的漲跌幅。有了這些之后，我們基本就可以開始著手撰寫程式了。
首先創建一個環境類：
times = [935, 940, 945, 950, 955, 1000, 1005, 1010, 1015,
1020, 1025, 1030, 1035, 1040, 1045, 1050, 1055,
1100, 1105, 1110, 1115, 1120, 1125, 1130, 1305,
1310, 1315, 1320, 1325, 1330, 1335, 1340, 1345,
1350, 1355, 1400, 1405, 1410, 1415, 1420, 1425,
1430, 1435, 1440, 1445, 1450, 1455, 1500]
class Market:
def __init__(self, data):
self.action_space = ['B', 'S', 'W'] # 買進、賣出、觀望
self.n_actions = len(self.action_space)
self.data = data # 935 940 ... 1500 48根K線的資料
self.time = 935
pass
def step(self, action):
# 要知道當前在那個狀態即時間點,用下一時間點的R（收益）作為
# 當前采取action的reward
tix = times.index(self.time)
nix = tix + 1
if self.time == 1500:
reward = 0
done = True
s_ = 'terminal'
# print('time is over.')
else:
reward = self.data.R.iloc[nix]
done = False
s_ = times[nix]
if action == 'B':
pass
elif action == 'S':
# 當R為-的時候，選擇S，應該是正獎勵
reward = reward * -1
else:
# 選擇觀望，既不虧損也不會盈利，但會損失機會成本
# 我們當前對觀望的決策持客觀態度，reward=0，這
# 可能需要在不同的大盤行情下適時調整
reward = 0
pass
self.time = s_
return s_, reward, done
pass
def reset(self):
self.time = 935
return self.time
pass
然后創建Q-learning演算法類（或者稱這個類為一個Agent）：
class QLearning:
#Agent
def __init__(self, actions, q_table=None, learning_rate=0.01,
discount_factor=0.9, e_greedy=0.1):
self.actions = actions # action 串列
self.lr = learning_rate # 學習速率
self.gamma = discount_factor # 折扣因子
self.epsilon = e_greedy # 貪婪度
# 列是action。
if q_table is None:
self.q_table = pd.DataFrame(columns=self.actions, dtype=np.float32) # Q 表
else:
self.q_table = q_table
# 檢測 q_table 中有沒有這個 state
# 如果還沒有當前 state, 那我們就插入一組全 0 資料, 作為這個 state 的所有 action 的初始值
def check_state_exist(self, state):
# state對應每一行，如果不在Q表中。
if state not in self.q_table.index:
# 插入一組全 0 資料，給每個action賦值為0
self.q_table = self.q_table.append(
pd.Series(
[0] * len(self.actions),
index=self.q_table.columns,
name=state,
)
)
# 根據 state 來選擇 action
def choose_action(self, state):
self.check_state_exist(state) # 檢測此 state 是否在 q_table 中存在
# 選行為，用 Epsilon Greedy 貪婪方法
if np.random.uniform() < self.epsilon:
# 隨機選擇 action
action = np.random.choice(self.actions)
else: # 選擇 Q 值最高的 action
state_action = self.q_table.loc[state, :]
# 同一個 state, 可能會有多個相同的 Q action 值, 所以我們亂序一下
state_action = state_action.reindex(np.random.permutation(state_action.index))
# 每一行中取到Q值最大的那個
action = state_action.idxmax()
return action
# 學習。更新 Q 表中的值
def learn(self, s, a, r, s_):
# s_是下一個狀態
self.check_state_exist(s_) # 檢測 q_table 中是否存在 s_
# Q(S,A) <- Q(S,A)+a*[R+v*max(Q(S',a))-Q(S,A)]
q_predict = self.q_table.loc[s, a] # 根據 Q 表得到的 估計（predict）值
# q_target 是現實值
if s_ != 'terminal': # 下個 state 不是 終止符
q_target = r + self.gamma * self.q_table.loc[s_, :].max()
else:
q_target = r # 下個 state 是 終止符
# 更新 Q 表中 state-action 的值
self.q_table.loc[s, a] += self.lr * (q_target - q_predict)
最后就是創建一個檔案來協調上面兩個類開始作業：
def update(data, q_table=None):
env = Market(data)
RL = QLearning(actions=env.action_space, q_table=q_table)
for episode in range(100):
# 初始化 state（狀態）
state = env.reset()
step_count = 0 # 記錄走過的步數
while True:
# 更新可視化環境
# env.render()
# RL 大腦根據 state 挑選 action
action = RL.choose_action(str(state))
# 探索者在環境中實施這個 action, 并得到環境回傳的下一個 state, reward 和 done (是否到了1500)
state_, reward, done = env.step(action)
step_count += 1 # 增加步數
# 機器人大腦從這個過渡（transition） (state, action, reward, state_) 中學習
RL.learn(str(state), action, reward, str(state_))
# 機器人移動到下一個 state
state = state_
# 如果時間到了1500, 這回合就結束了，或者是某個止損條件達到了
if done:
# print("回合 {} 結束. 總步數 : {}\n".format(episode + 1, step_count))
break
# print('模擬交易結束了。')
# print('\nQ 表:')
# print(RL.q_table)
return RL.q_table
def train():
code = '000001' # 上證指數
sd = dt.datetime(2018, 10, 1)
ed = dt.datetime(2018, 11, 1)
# 我已經把從jqdata讀取到了資料存在了本地，這里只是讀取出來
data = md().read_data('index_min5', stock_code=code,
date={'gte': sd, 'lt': ed},
field={'_id': 0, 'time': 1, 'close': 1, 'date': 1})
data = data.sort_values(['date', 'time'], ascending=False)
# 計算每根K線收盤時未來三根K線的漲跌幅
data['R'] = (data.close.shift(3) / data.close - 1) * 100
data.fillna(0, inplace=True)
data = data.round({'R': 3})
data = data.sort_values(['date', 'time'], ascending=True)
qtb = None
for k, g in data.groupby(['date']):
print('train to:', k)
try:
# 開始一天一天的訓練
qtb = update(g, qtb)
except Exception as e:
ExceptionInfo(e)
print('\nQ 表:')
print(qtb)
qtb['time'] = qtb.index
qtb.to_csv(path_or_buf='E:\wv\ReinfL\model_param\qtb({})_{}.csv'.
format(code, sd.strftime('%Y_%m_%d')), index=False)
pass
train()
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/80756.html
上一篇：self.rect.centerx和self.center的轉換問題
下一篇：如何在C語言中嵌入匯編語言？