作为棋盘游戏，围棋已有2000多年的历史。长期以来，它一直被认为是美与艺术的创造，因为其规则很简单，但对弈非常复杂，并且曾认为在未来几十年中能经受住人工智能的挑战。然而，当时间来到2016年3月9日至15日，在韩国首尔进行的韩国围棋九段棋手李世石与人工智能围棋程序阿尔法围棋AlphaGo之间的五番棋比赛，AlphaGo以4-1战胜了这位世界顶级围棋棋手，这是人工智能里程碑式的成就。
AlphaGo是使用基于数百万人类专家游戏的训练数据集以及其他资源进行监督学习的。AlphaGo的下一个迭代版本AlphaGo Zero，完全跳过了这种方法，仅依靠强化学习和自我对弈，将不同代且训练有素的基于神经网络的人工智能体放在一起相互竞争。Silver等人在其2017年发表的文章中介绍了AlphaGo Zero的详细信息。在摘要中， 研究人员总结道:
AlphaGo成了自己的老师:训练神经网络来预测自己的走法选择以及游戏的获胜者。该神经网络提高了树搜索的强度，从而能在下一次迭代中产生更高质量的走 法选择和更强的自我对弈。从一无所知开始，新程序 AlphaGo Zero 取得了超人的表现，以100–0战胜了之前发布的击败了李世石的AlphaGo。
由此可见，在行动策略中将强化学习与神经网络相结合的优越性。下面我们来简单学习下“强化学习”。
基本概念
- 环境
环境定义了当前的问题，可以是要玩的计算机游戏或要进行交易的金融市场。
- 状态
状态包含描述环境当前状态的所有相关参数。在计算机游戏中，这可能是整个屏幕及其 像素。在金融市场中，这可能包括当前和历史价格水平或金融指标，比如移动平均线、 宏观经济变量等。
- 智能体
智能体这个词包含了与环境交互并从这些交互中学习的强化学习算法的所有元素。在游 戏环境中，智能体可能代表玩游戏的玩家。在金融环境中，智能体可以代表在市场中进 行交易的交易者。
- 动作
智能体可以从一组(有限的)被允许的动作中选择一个动作。在计算机游戏中，被允许 的动作可能是向左或向右移动，而在金融市场中，被允许的动作可能是做多或做空。
- 步骤
给定智能体的动作，环境状态会被更新，这样的更新通常被称为一个步骤。步骤的概念 可以包含两个步骤之间的相同或者不同的时间间隔。虽然在计算机游戏中，与游戏环境 的实时交互是通过相当短且相同的时间间隔(“游戏时钟”)来模拟的，但诸如与金融市 场环境交互的交易机器人则可以在更长且不同的时间间隔内采取动作。
- 奖励
根据智能体选择的动作，对其实行奖励(或惩罚)。对于计算机游戏，积分是一种典型 的奖励。在金融环境中，利润(或亏损)是一种标准的奖励(或惩罚)。
- 目标
目标是指智能体试图最大化的内容。在计算机游戏中，这通常是智能体达到的分数。对 于金融交易机器人，这可能是累积的交易利润。
- 策略
策略定义了智能体在给定环境状态下所采取的动作。给定计算机游戏的特定状态(由构 成当前场景的所有像素表示)，策略可能会指定智能体选择“向右移动”作为动作。观 察到连续 3 个价格上涨的交易机器人可能会根据其策略决定做空市场。
- 回合
回合是从环境的初始状态到成功或可预见的失败的一组步骤。在游戏中，这是从游戏开 始到输或赢为止。以金融界为例，这就是从年初到年底，或者到破产。
接下来我们在实例中逐步说明上述概念。
OpenAI Gym
OpenAI是一个致力于促进人工智能研究，特别是强化学习研究的组织。 OpenAI开发并开源了一套环境，称为 OpenAI Gym，允许通过标准化API训练RL智能体。在众多环境中，有模拟经典强化学习问题的CartPole环境(或游戏)，即把一根杆子直立在推车上，目的是通过左右移动推车来学习平衡杆子的策略。环境状态由4个参数表示， 包括以下物理测量值：推车位置、推车速度、极角和极角速度(尖端)。图1描述了一个可视化的环境。
图1 OpenAI Gym的CartPole环境
以下代码实例化CartPole的环境对象，观察空间是环境状态的模型。
env = gym.make('CartPole-v1', render_mode="rgb_array")
env.action_space.seed(100)
# 初始状态:推车位置、推车速度、极角和极角速度
state, _ = env.reset()
state
# Out:
# array([ 0.0438, -0.0123, 0.0364, -0.0128], dtype=float32) 只要 done=False，智能体就还在游戏中，并且可以选择另一个动作。当智能体达到总计500步(CartPole-v0版本是200步)或总奖励达到500(每步奖励 1.0)时，即取得成功。当推车上的杆子到达一定角度导致杆子从推车上掉下来时，故障被观测到，在这种情况下，返回done=True。简单智能体是遵循完全随机策略的智能体：无论观察到什么状态，智能体都会选择一个随机动作，如下面的代码实现所示。在这种情况下，智能体可以走的步数仅取决于它的幸运程度，其并未以更新策略的形式进行学习。
env.reset()
for e in range(500):
# 随机动作策略
a = env.action_space.sample()
# 向前一步
state, reward, done, truncated, info = env.step(a)
print(f'step={e:2d} | state={state} | action={a} | reward={reward}')
# 少于200步则失败
if done and (e + 1) < 200:
print('*** FAILED ***')
break
# Out:
# step= 1 | state=[ 0.0096 -0.1469 -0.0307 0.2928] | action=0 | reward=1.0
# step= 2 | state=[ 0.0066 -0.3415 -0.0249 0.5757] | action=0 | reward=1.0
# step= 3 | state=[-0.0002 -0.5363 -0.0133 0.8604] | action=0 | reward=1.0
# step= 4 | state=[-0.0109 -0.7312 0.0039 1.1489] | action=0 | reward=1.0
# step= 5 | state=[-0.0255 -0.9264 0.0268 1.4428] | action=0 | reward=1.0
# step= 6 | state=[-0.0441 -1.1218 0.0557 1.7437] | action=0 | reward=1.0
# step= 7 | state=[-0.0665 -1.3175 0.0906 2.0532] | action=0 | reward=1.0
# step= 8 | state=[-0.0928 -1.5135 0.1316 2.3725] | action=0 | reward=1.0
# step= 9 | state=[-0.1231 -1.7095 0.1791 2.7025] | action=0 | reward=1.0
# step=10 | state=[-0.1573 -1.9054 0.2331 3.0441] | action=0 | reward=1.0
# *** FAILED ***
done
# Out:
# True 在监督学习中，假设训练数据集、验证数据集和测试数据集在训练开始之前已经存在，而在强化学习中，智能体通过与环境交互来生成自己的数据。在许多情况下(比如在游戏中)，这是一个巨大的简化。考虑一下国际象棋游戏:一个 RL智能体可以通过与另一个国际象棋引擎或另一个版本的自身对战，而不是将数千个人类历史上所生成的国际象棋棋谱加载到计算机中。
蒙特卡罗智能体
CartPole问题不一定非要使用成熟的强化学习方法和一些神经网络来解决，下面是基于蒙特卡罗模拟的简单解决方案，并使用降维的特定策略。在这种情况下，定义环境状态的4个参数通过线性组合被压缩为了单个实值参数。
np.random.seed(100)
weights = np.random.random(4) * 2 - 1
# weights
print(f'weights: {weights}')
# 环境的初始状态
state, _ = env.reset()
print(f'state: {state}')
# 状态和权重的点积
s = np.dot(state, weights)
print(f's: {s}')
# Out:
# weights: [ 0.0868 -0.4433 -0.151 0.6896]
# state: [-0.0465 -0.0319 -0.0016 -0.0308]
# s: -0.010908021653158767 接下来可以使用此策略玩一回合CartPole游戏。鉴于所应用的权重的随机性，通常结果并不比上一节的随机动作策略的结果好。
def run_episode(env, weights):
state = env.reset()[0]
treward = 0
for _ in range(500):
s = np.dot(state, weights)
# 根据单个状态参数s的符号来选择策略
a = 0 if s < 0 else 1
state, reward, done, truncated, info = env.step(a)
treward += reward
if done:
break
return treward
# run_episode(env, weights) 因此，可以应用蒙特卡罗模拟来测试大量不同的权重，检查它们是成功还是失败，然后选择产生成功的权重。
def set_seeds(seed=100):
random.seed(seed)
np.random.seed(seed)
set_seeds()
num_episodes = 1000
besttreward = 0
for e in range(1, num_episodes + 1):
# 随机权重
weights = np.random.rand(4) * 2 - 1
# 这些权重的总奖励
treward = run_episode(env, weights)
# 观察是否有改善
if treward > besttreward:
# 替换最佳总奖励
besttreward = treward
# 替换最佳权重
bestweights = weights
if treward == 200:
print(f'SUCCESS | episode={e}')
break
print(f'UPDATE | episode={e}')
# Out:
# UPDATE | episode=1
# UPDATE | episode=2
# UPDATE | episode=13
# UPDATE | episode=35 如果连续 100 回合的平均总奖励为195或更高，则认为CartPole问题已被智能体解决，如以下代码所示。
res = []
for _ in range(100):
treward = run_episode(env, bestweights)
res.append(treward)
res[:10]
# Out:
# [500.0, 500.0, 500.0, 500.0, 500.0, 500.0, 500.0, 500.0, 500.0, 500.0] 这是一个用来与其他更复杂的方法做比较的强有力的基准。
神经网络智能体
CartPole游戏也可以被转换为一个分类问题：环境状态由4个特征值组成，对应每组给定特征值的正确操作是需要预测的标签。通过与环境交互，神经网络智能体可以收集由特征值和标签组合组成的数据集。给定这个不断增长的数据集，可以训练神经网络来学 习给定环境状态的正确动作。在这种情况下，神经网络代表策略，智能体会根据新体验更新策略。
首先是一些库的引用。
import tensorflow as tf
from keras.layers import Dense, Dropout
from keras.models import Sequential
from keras.optimizers import adam_v2, rmsprop_v2
from sklearn.metrics import accuracy_score
def set_seeds(seed=100):
random.seed(seed)
np.random.seed(seed)
tf.random.set_seed(seed)
env.action_space.seed(100) 然后是一个NNAgent类，它结合了智能体的主要元素:策略的神经网络模型、根据策略选择动作、更新策略(训练神经网络)，以及多个回合的学习过程本身。智能体同时使用探索和利用来选择一个动作。探索是指随机动作，独立于当前策略。利用是指从当前策略派生的动作。这么做是因为某种程度的探索可以确保获得更丰富的经验，从而使智能体的学习得到改善。
class NNAgent:
def __init__(self):
# 最高总奖励
self.max = 0
self.scores = list()
self.memory = list()
self.model = self._build_model()
# 策略的 DNN 分类模型
def _build_model(self):
model = Sequential()
model.add(Dense(24, input_dim=4,
activation='relu'))
model.add(Dense(1, activation='sigmoid'))
model.compile(loss='binary_crossentropy', optimizer=rmsprop_v2.RMSprop(learning_rate=0.001))
return model
# 选择动作的方法(探索和利用)
def act(self, state):
if random.random() <= 0.5:
return env.action_space.sample()
action = np.where(self.model.predict(
state, batch_size=None)[0, 0] > 0.5, 1, 0)
return action
# 更新策略的方法(训练神经网络)
def train_model(self, state, action):
self.model.fit(state, np.array([action,]),
epochs=1, verbose=False)
# 从与环境的交互中学习的方法
def learn(self, episodes):
for e in range(1, episodes + 1):
state = env.reset()[0]
for _ in range(500):
state = np.reshape(state, [1, 4])
action = self.act(state)
next_state, reward, done, truncated, info = env.step(action)
if done:
score = _ + 1
self.scores.append(score)
self.max = max(score, self.max)
print('episode: {:4d}/{} | score: {:3d} | max: {:3d}'
.format(e, episodes, score, self.max), end='\r')
break
self.memory.append((state, action))
self.train_model(state, action)
state = next_state 神经网络智能体没有解决上述配置下的问题，最大总奖励甚至一次也没有达到200。
set_seeds(100)
agent = NNAgent()
episodes = 500
agent.learn(episodes)
# Out:
# episode: 500/500 | score: 12 | max: 45
# 所有回合的平均总奖励
sum(agent.scores) / len(agent.scores)
# Out:
# 13.89 这种方法似乎缺少一些东西，而其中一个主要的缺失元素是超越当前状态和要选择的动作的想法。目前为止所实现的方法无论如何都没有考虑到只有当智能体连续存活200步时才 能取得成功。简单地说，智能体会避免采取错误的动作，但没有学会赢得比赛。
分析收集到的状态(特征)和动作(标签)的历史数据表明，神经网络的准确率达到了75%左右。然而，这并没有转化为之前看到的获胜策略。
# 所有回合的特征(状态)
f = np.array([m[0][0] for m in agent.memory])
# 所有回合的标签(动作)
l = np.array([m[1] for m in agent.memory])
accuracy_score(np.where(agent.model.predict(f) > 0.5, 1, 0), l)
# Out:
# 0.7453840186190845 DQL智能体
Q学习是一种强化学习算法，除了会考虑来自动作的即时奖励，还会考虑延迟奖励。该算法归功于Watkins(1989)以及Watkins 和 Dayan(1992)，并在 Sutton 和 Barto(2018)的第6章中有详细解释。Q学习解决了神经网络智能体遇到的超越下一个立即奖励的问题。
该算法的工作原理大致如下：有一个动作–价值策略Q，它为每个状态和动作的组合分配一个值。值越高，从智能体的角度来看动作越好。如果智能体使用策略Q选择一个动作， 则它会选择具有最高值的动作。
那么一个动作的价值是如何得出的呢？一个动作的价值由它的直接奖励和下一状态下最优动作的折现值组成，以下是正式表达。
这里，是步骤(时间)t的状态， 是状态采取的动作，是动作的直接奖励，是折扣因子， 是给定当前策略 Q 的最优动作下的最大延迟奖励。 a在一个只有有限数量的可能状态的简单环境中，Q 可以用一张表格表示，其中列出了每个状态–动作组合的相应值。然而，在更有趣或更复杂的设置中(如CartPole环境)，状态数量过多，无法通过穷尽的方式写出Q，因此，Q通常被理解为一个函数。
这就是神经网络发挥作用的地方。在现实的设置和环境中，函数Q的封闭形式可能不存在，或者利用动态规划方法很难推导。因此Q学习算法通常仅以近似值为目标。神经网络具有通用逼近能力，是完成Q函数逼近的自然选择。
Q学习的另一个关键要素是重放。DQL智能体会重放许多经验(状态–动作组合)以定期更新策略函数Q，这可以大大提高学习效果。此外，下面介绍的DQL智能体(DQLAgent) 于学习过程中在探索和利用之间交替。交替会以系统的方式进行，因为智能体仅从探索 开始(一开始它不可能学到任何东西)，然后会缓慢但稳定地降低探索率 ϵ 直到达到最低水平。
from collections import deque
from keras.optimizers import adam_v2, rmsprop_v2
class DQLAgent:
def __init__(self, gamma=0.95, hu=24, opt=adam_v2.Adam, lr=0.001, finish=False):
self.finish = finish
# 初始探索率
self.epsilon = 1.0
# 最小探索率
self.epsilon_min = 0.01
# 探索率的衰减率
self.epsilon_decay = 0.995
# 延迟奖励的折扣因子
self.gamma = gamma
# 重放的批次大小
self.batch_size = 32
self.max_treward = 0
self.averages = list()
# 有限历史的双端队列 deque 集合
self.memory = deque(maxlen=2000)
self.osn = env.observation_space.shape[0]
self.model = self._build_model(hu, opt, lr)
def _build_model(self, hu, opt, lr):
model = Sequential()
model.add(Dense(hu, input_dim=self.osn,
activation='relu'))
model.add(Dense(hu, activation='relu'))
model.add(Dense(env.action_space.n, activation='linear'))
model.compile(loss='mse', optimizer=opt(learning_rate=lr))
return model
def act(self, state):
if random.random() <= self.epsilon:
return env.action_space.sample()
action = self.model.predict(state)[0]
return np.argmax(action)
def replay(self):
# 随机选择历史批次进行回放。
batch = random.sample(self.memory, self.batch_size)
for state, action, reward, next_state, done in batch:
if not done:
reward += self.gamma * np.amax(
self.model.predict(next_state)[0]) # 状态–动作对的Q值
target = self.model.predict(state)
target[0, action] = reward
# 为新的动作–价值对更新神经网络
self.model.fit(state, target, epochs=1,
verbose=False)
if self.epsilon > self.epsilon_min:
# 更新探索率
self.epsilon *= self.epsilon_decay
def learn(self, episodes):
trewards = []
for e in range(1, episodes + 1):
state, _ = env.reset()
state = np.reshape(state, [1, self.osn])
for _ in range(5000):
action = self.act(state)
# next_state, reward, done, info = env.step(action)
next_state, reward, done, truncated, info = env.step(action)
next_state = np.reshape(next_state,
[1, self.osn])
# 存储新数据
self.memory.append([state, action, reward,
next_state, done])
state = next_state
if done:
treward = _ + 1
trewards.append(treward)
av = sum(trewards[-25:]) / 25
self.averages.append(av)
self.max_treward = max(self.max_treward, treward)
templ = 'episode: {:4d}/{} | treward: {:4d} | '
templ += 'av: {:6.1f} | max: {:4d}'
print(templ.format(e, episodes, treward, av,
self.max_treward), end='\r')
break
if av > 195 and self.finish:
print()
break
if len(self.memory) > self.batch_size:
# 根据过去的经验重放以更新策略
self.replay()
def test(self, episodes):
trewards = []
for e in range(1, episodes + 1):
state = env.reset()
for _ in range(5001):
state = np.reshape(state, [1, self.osn])
action = np.argmax(self.model.predict(state)[0])
next_state, reward, done, truncated, info = env.step(action)
state = next_state
if done:
treward = _ + 1
trewards.append(treward)
print('episode: {:4d}/{} | treward: {:4d}'
.format(e, episodes, treward), end='\r')
break
return trewards DQL智能体如何执行？如以下代码所示，它达到了CartPole的获胜状态，总奖励为200。 图2显示了分数的移动平均线以及它是如何随时间增加的，尽管不是单调增加。相反， 智能体的性能有时会显著下降。除此之外，一直在进行的探索会导致随机操作，这不一定会导致总奖励方面的良好结果，但可能会为更新策略网络带来有益的经验。
episodes = 1000
set_seeds(100)
agent = DQLAgent(finish=True)
agent.learn(episodes)
# Out:
# episode: 510/1000 | treward: 224 | av: 199.4 | max: 740
plt.figure(figsize=(10, 6))
x = range(len(agent.averages))
y = np.polyval(np.polyfit(x, agent.averages, deg=3), x)
plt.plot(agent.averages, label='moving average')
plt.plot(x, y, 'r--', label='trend')
plt.xlabel('episodes')
plt.ylabel('total reward')
plt.legend(); 
图2 CartPole的DQLAgent平均总奖励
DQL智能体是否解决了CartPole问题？在现有的游戏设置下，考虑到OpenAI Gym对成功的定义，确实解决了。
trewards = agent.test(100)
# Out:
# episode: 100/100 | treward: 468
sum(trewards) / len(trewards)
# Out:
# 532.79 简单的金融沙箱
为了将Q学习方法迁移至金融领域，下面展示了一个模拟OpenAI Gym环境的类，但其仅适用于以金融时间序列数据为代表的金融市场。这里的思路是，类似于CartPole环境，4个历史价格代表金融市场的状态，当给定一个状态时，智能体可以决定是做多还是做空。 在这种情况下，两个环境是可比较的，因为每一个状态都是由4个参数给出，并且智能体可以采取两种不同的动作。
为了模仿 OpenAI Gym API，需要两个辅助类，一个用于观察空间，一个用于动作空间。
class observation_space:
def __init__(self, n):
self.shape = (n,)
class action_space:
def __init__(self, n):
self.n = n
def seed(self, seed):
pass
def sample(self):
return random.randint(0, self.n - 1) 以下代码定义了Finance类，该类提取了许多交易品种的日终历史价格。该类的主要方法是 .reset() 和 .step()，其中，.step()方法会检查是否采取了正确的动作，相应地定义奖励，并检查成功或失败。当智能体能够通过整个数据集正确交易时，就取得了成功。当然，这可以有不同的定义(例如，定义当智能体成功交易1000步时就取得了成功)。失败被定义为准确率小于 50%(总奖励除以总步数)。然而，这仅在一定数量的步骤之后进行检查，以避免该指标的初始方差过高。
class Finance:
url = 'http://hilpisch.com/aiif_eikon_eod_data.csv'
def __init__(self, symbol, features):
self.symbol = symbol
self.features = features
self.observation_space = observation_space(4)
self.osn = self.observation_space.shape[0]
self.action_space = action_space(2)
# 定义所需的最低准确率
self.min_accuracy = 0.475
self._get_data()
self._prepare_data()
def _get_data(self):
self.raw = pd.read_csv(self.url, index_col=0,
parse_dates=True).dropna()
def _prepare_data(self):
self.data = pd.DataFrame(self.raw[self.symbol])
self.data['r'] = np.log(self.data / self.data.shift(1))
self.data.dropna(inplace=True)
self.data = (self.data - self.data.mean()) / self.data.std()
self.data['d'] = np.where(self.data['r'] > 0, 1, 0)
def _get_state(self):
# 选择用于定义金融市场状态的数据
return self.data[self.features].iloc[self.bar - self.osn:self.bar].values
def seed(self, seed=None):
pass
# 将环境重置为其初始值
def reset(self):
self.treward = 0
self.accuracy = 0
self.bar = self.osn
state = self.data[self.features].iloc[
self.bar - self.osn:self.bar]
return state.values
def step(self, action):
# 检查智能体是否选择了正确的动作(成功交易)
correct = action == self.data['d'].iloc[self.bar]
# 定义智能体收到的奖励
reward = 1 if correct else 0
# 将奖励添加到总奖励中
self.treward += reward
# 使环境向前运行一步
self.bar += 1
# 计算给定所有步骤(交易)的成功的动作(交易)的准确率
self.accuracy = self.treward / (self.bar - self.osn)
# 如果智能体到达数据集的末尾，则成功
if self.bar >= len(self.data):
done = True
elif reward == 1:
# 如果智能体采取了正确的动作，那么它可以继续前进
done = False
elif (self.accuracy < self.min_accuracy and self.bar > self.osn + 10):
# 如果在一些初始步骤之后，准确率下降到最低水平以下，则该回合结束(失败)
done = True
else:
# 对于其余情况，智能体可以继续前进
done = False
state = self._get_state()
info = {}
return state, reward, done, info Finance 类的实例表现得很像OpenAI Gym的环境。特别是，在这种基本情况下，该实例表现得与 CartPole 环境完全一样。
# 指定用于定义代表状态数据的交易标的代号和特征类型(交易标的代号或对数收益率)
env = Finance('EUR=', 'EUR=')
env.reset()
# Out:
# array([1.819 , 1.8579, 1.7749, 1.8579])
a = env.action_space.sample()
a
# Out:
# 0
env.step(a)
# Out:
# (array([1.8579, 1.7749, 1.8579, 1.947 ]), 0, False, {}) 虽然为CartPole游戏开发的DQLAgent能学习在金融市场中进行的交易，但结果并不“惊艳”(参见图3)。
set_seeds(100)
agent = DQLAgent(gamma=0.5, opt=rmsprop_v2.RMSprop)
episodes = 1000
agent.learn(episodes)
# Out:
# episode: 1000/1000 | treward: 2511 | av: 1821.2 | max: 2511
agent.test(3)
# Out:
# episode: 3/3 | treward: 2511
# [2511, 2511, 2511]
plt.figure(figsize=(10, 6))
x = range(len(agent.averages))
y = np.polyval(np.polyfit(x, agent.averages, deg=3), x)
plt.plot(agent.averages, label='moving average')
plt.plot(x, y, 'r--', label='regression')
plt.xlabel('episodes')
plt.ylabel('total reward')
plt.legend(); 图3 运行于Finance环境的DQLAgent的平均总奖励
通过模拟OpenAI Gym环境的API，我们将DQL智能体应用于金融市场环境，无须对智能体本身进行任何更改。尽管智能体在这种新环境中的表现可能并不优异，但它说明了强化学习方法是相当通用的。