機械学習と戯れる：Q-Learningでマーケティング系のActionを行うべきかの判断ができるか？

はじめに

以前 足し算ゲームを強化学習で学習できるか？ を試してみて、問題なく学習ができました。

今回はもう少し現実的な問題を想定してみようと思います。

• Webサイトに来るユーザに対して、Webサイト運営者が「あるアクション(メール？クーポン？など）」を起こすと、望ましい行動(そのユーザが何か購入するなど)を取る、とします
• その時、どのユーザにどのタイミングでどのアクションを取ればいいか、を知りたい

という問題があります。
まあ、メールぐらいなら全員に送れば良いじゃん的な話はありますが、送りすぎると離脱に繋がりますし、クーポンはコストもかかるのであまり乱発したくはないです。

この問題を Q-Learning的な枠組みでやったらどうなるのだろうか、というのが今回のお題です。
Q-Learningだと、アクションが複数になっても対応できるのが良い所です。
といっても、簡単な完全に仮想的なシチュエーションで考えます。

お題：Q-Learningでマーケティング系のActionを行うべきかの判断ができるか？ ゲーム

ゲーム

簡単に書くと

• ユーザの行動Uは 0~3
• 運営者のアクションAは 0~1
• U=2 のときに A=1 を行うと良い
• ただし報酬はすぐには得られれず、4ターン後に+1を得られる
• U=2じゃないときに、A=1を行うとペナルティ
• 状態Sは (U,A)の5回分の履歴

という感じです。

何故こんなゲームを考えたのか

このゲームの場合、「最適な行動をとってから報酬を貰えるまでに間隔がある」というのが難しいのではないかと心配になりました。最適な行動の直後に報酬が貰えれば学習が早そうなのですが、そうではないところが最大のポイントです。本当は「ペナルティ」の定義も難しいのかなと思いますが、今は単純に間違ったっぽいアクションに対してすぐに与えています。（これだと問題が簡単すぎてしまうのか・・・？）。

ただ、強化学習で迷路を探索する にもあるように、遡って学習することができるので、できるのだろうとは思いつつもちょっと検証してみたかったというところです。
※ ちなみに上記の強化学習の迷路を学習していく動画は面白いので興味があれば是非みると良いと思います！

結果

報酬が貰えるのは、U=2が発生した時のみですので、U=2が発生した時に chance_count というのを増やします。ある一定期間の間に得られる報酬の最大は chance_count になります。
そこで、 得られた報酬 / chance_count を hit_rate として、
10000回の学習・評価毎にどう変化したかを以下の図に示します。

5000万回ほどやってみましたが、3000万回くらいで学習は頭打ちになっている感じで、およそ hit_rate=0.9 くらいでした。

考察

• 一応良いアクションを学習していくことがわかりました。問題を複雑にしていって、どこまで追随できるのかは今後の課題になりそうです。
• 割りと簡単なルールなのに 100% にならないのが少し意外でした
  • 局所解？？バグ？？
• 学習回数も結構多いなぁ、と。状態数は U(4)*A(2)^HISTORY(5) -> 32768 くらいですが、、、でもまあ、そんなものなのかな。。
• 実際にはDeep Q Learning的に行わないと少なくとも実用性はなさそうですが、DeepLearning的にやると学習に更なる時間がかかるので、何かうまい工夫が必要そうです
• 実際にはペナルティがもう少しわかりにくい可能性がある？
  • 例えば、ユーザが離脱するというのはずーっとU=0が続くということですが、Aとの因果関係は本当によくわからない
  • でもそれ以外は、Uのマイナスな行動（退会、悪口、クレーム）やAのコスト（費用）などとすればわかりやすいといえばわかりやすいのか。

さいごに

Q Learningはある意味単純なAIっぽいので楽しいです。
何かに使えるといいんだけどなぁ。

コード

参考までに貼っておきます。

#!/usr/bin/env python
# coding: utf-8

import numpy as np
from random import random, choice


class Game(object):
    state = None
    actions = None
    game_over = False

    def __init__(self, player):
        self.player = player
        self.turn = 0
        self.last_reward = 0
        self.total_reward = 0
        self.init_state()

    def player_action(self):
        action = self.player.action(self.state, self.last_reward)
        if action not in self.actions:
            raise Exception("Invalid Action: '%s'" % action)
        self.state, self.last_reward = self.get_next_state_and_reward(self.state, action)

    def play(self):
        yield(self)
        while not self.game_over:
            self.player_action()
            self.turn += 1
            self.total_reward += self.last_reward
            yield(self)

    def init_state(self):
        raise NotImplemented()

    def get_next_state_and_reward(self, state, action):
        raise NotImplemented()


class UserAndPushEventGame(Game):
    """
    State           :S : list of (U, A)
    UserActivity    :U : int of 0~3
    Action          :A : int of 0 or 1

    Next-State(S, A):S':
        S[-1][1] = A
        S.append((Next-U, None))
        S = S[-5:]
    Next-U          :  :
        if S[-4] == (2, 1) then 3
        else 10% -> 2, 10% -> 1, 80% -> 0
    Reward(S, A)    :R :
        if S[-1] == (3, *) then R += 1
        wrong_action_count := Number of ({0,1,3}, 1) in S
        R -= wrong_action_count * 0.3
    """

    STATE_HISTORY_SIZE = 5

    def init_state(self):
        self.actions = [0, 1]
        self.state = [(0, None)]
        self.chance_count = 0

    def get_next_state_and_reward(self, state, action):
        next_state = (state + [(self.next_user_action(state), None)])[-self.STATE_HISTORY_SIZE:]
        next_state[-2] = (next_state[-2][0], action)
        reward = 0
        if len(state) > 0 and state[-1][0] == 3:
            reward += 1
        action_count = reduce(lambda t, x: t+(x[1] or 0), state, 0)
        correct_action_count = len([0 for x in state if x == (2, 1)])
        wrong_action_count = action_count - correct_action_count
        reward -= wrong_action_count * 0.3
        return next_state, reward

    def next_user_action(self, state):
        if len(state) > 4 and state[-4] == (2, 1):
            return 3
        else:
            rnd = np.random.random()
            if rnd < 0.8:
                return 0
            elif rnd < 0.9:
                return 1
            else:
                self.chance_count += 1
                return 2


class HumanPlayer(object):
    training = False

    def action(self, state, last_reward):
        print "LastReward=%s, CurrentState: %s" % (last_reward, state)
        while True:
            action_input = raw_input("Enter 0~1: ")
            if int(action_input) in [0, 1]:
                return int(action_input)


class QLearnPlayer(object):
    ALPHA = 0.1
    GAMMA = 0.99
    E_GREEDY = 0.05

    def __init__(self):
        self.actions = [0, 1]
        self.q_table = {}
        self.last_state = self.last_action = None
        self.training = True

    def get_q_value(self, state, action):
        return self.q_table.get(state, {}).get(action, (np.random.random() - 0.5)/1000)  # 未定義は小さい乱数を返す

    def get_all_q_values(self, state):
        return [self.get_q_value(state, act) for act in self.actions]

    def set_q_value(self, state, action, val):
        if state in self.q_table:
            self.q_table[state][action] = val
        else:
            self.q_table[state] = {action: val}

    def action(self, state, last_reward):
        state = tuple(state)
        next_action = self.select_action(state)
        if self.last_state is not None:
            self.update_q_table(self.last_state, self.last_action, state, last_reward)
        self.last_state = state
        self.last_action = next_action
        return next_action

    def select_action(self, state):
        if self.training and random() < self.E_GREEDY:
            return choice(self.actions)
        else:
            return np.argmax(self.get_all_q_values(state))

    def update_q_table(self, last_state, last_action, cur_state, last_reward):
        if self.training:
            d = last_reward + np.max(self.get_all_q_values(cur_state)) * self.GAMMA - self.get_q_value(last_state, last_action)
            self.set_q_value(last_state, last_action, self.get_q_value(last_state, last_action) + self.ALPHA * d)


if __name__ == '__main__':
    SWITCH_MODE_TURN_NUM = 10000
    fp = file("result.txt", "w")
    dt = file("detail.txt", "w")
    player = QLearnPlayer()
    # player = HumanPlayer()
    game = UserAndPushEventGame(player)
    last_chance_count = last_score = 0
    for g in game.play():
        # dt.write("%s: isT?=%s LastReward=%s TotalReward=%s S=%s\n" %
        #          (g.turn, player.training, g.last_reward, g.total_reward, g.state))
        if g.turn % SWITCH_MODE_TURN_NUM == 0:
            if not player.training:
                this_term_score = game.total_reward - last_score
                this_term_chance = game.chance_count - last_chance_count
                if this_term_chance > 0:
                    hit_rate = 100.0*this_term_score/this_term_chance
                else:
                    hit_rate = 0
                # print "Turn=%d: This 100 turn score=%2.2f chance=%02d: HitRate=%.1f%% %s" % \
                #       (g.turn, this_term_score, this_term_chance, hit_rate, '*' * int(hit_rate/2))
                fp.write("%d\t%.2f\t%d\t%f\n" % (g.turn, this_term_score, this_term_chance, hit_rate))
            last_score = game.total_reward
            last_chance_count = game.chance_count
            player.training = not player.training
        if g.turn % 10000 == 0:
            fp.flush()