日経平均株価の時系列データの取得

時系列データの取得その1

日経平均株価の過去のcsvデータを取ってきて保存します。

# csvを読み込む時のコード
import pandas as pd
from io import StringIO
import urllib

def read_csv(url):
    res = urllib.request.urlopen(url)
    res=res.read().decode('shift_jis')
    df = pd.read_csv(StringIO( res) )
    return df

# 日経平均株価の時系列データの取得
import pandas as pd
from io import StringIO
import urllib

# 上記の関数を使い日経平均株価の時系列データを取得
url = "https://indexes.nikkei.co.jp/nkave/historical/nikkei_stock_average_daily_jp.csv"
def read_csv(url):
    res = urllib.request.urlopen(url)
    res = res.read().decode('shift_jis')
    df = pd.read_csv(StringIO(res))
    # 必要のない最後の行を取り除いています
    df = df.drop(df.shape[0]-1)
    return df

# dfというdataframeで保存し、出力
df = read_csv(url)
# 出力
df

時系列データの取得その2

時系列データの終値に注目して予測を行います。

# 時系列データの取得その1,日経平均株価の時系列データの取得コードを載せます。
# indexを日付にした後、時系列にします
df["データ日付"] = pd.to_datetime(df["データ日付"], format='%Y/%m/%d')
df = df.set_index('データ日付')

# カラムから'始値', '高値', '安値'を取り除いて、日付が古い順に並べます
df = df.drop(['始値', '高値', '安値'], axis=1)
df = df.sort_index(ascending=True)
df

時系列データの取得その3

import pandas as pd

# dfとdf_tweetsの二つのテーブルをindexをキーとして結合し、Nanを消去
df_tweets = pd.read_csv('./6050_stock_price_prediction_data/df_tweets.csv', index_col='date')
table = df_tweets.join(df, how='right').dropna()
# table.csvとして出力
table.to_csv("./6050_stock_price_prediction_data/table.csv")
table

株価予測

株価予測その1

Python: 株価予測その①で説明したように
今回はテクニカル分析を用いて株価の予測を行います。

過去三日間の日経平均株価の時系列の変化とPN値の変化を特徴量にして
次の日の株価の上下の予測を行います。

訓練データとテストデータの二つに分け、訓練データを標準化したのち、訓練データの平均と分散を用いてテストデータの標準化を行います。

# 基本雛形
from sklearn.model_selection import train_test_split

X = table.values[:, 0]
y = table.values[:, 1]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0, shuffle=False)
X_train_std = (X_train - X_train.mean()) / X_train.std()
X_test_std = (X_test - X_train.mean()) / X_train.std()

# 学習データの作成
from sklearn.model_selection import train_test_split

X = table.values[:, 0]
y = table.values[:, 1]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0, shuffle=False)
X_train_std = (X_train - X_train.mean()) / X_train.std()
X_test_std = (X_test - X_train.mean()) / X_train.std()

# df_trainというテーブルを作りそこにindexを日付、カラム名をpn値、終値にしてdf_train.csvという名前でdataフォルダ内に出力
df_train = pd.DataFrame(
    {'pn': X_train_std,
     '終値': y_train},
    columns=['pn', '終値'],
    index=table.index[:len(X_train_std)])
df_train.to_csv('./6050_stock_price_prediction_data/df_train.csv')

# テストデータについても同様にdf_testというテーブルを作り、df_test.csvという名前でdataフォルダ内に出力
df_test = pd.DataFrame(
    {'pn': X_test_std,
     '終値': y_test},
    columns=['pn', '終値'],
    index=table.index[len(X_train_std):])
df_test.to_csv('./6050_stock_price_prediction_data/df_test.csv')

株価予測その2

# まず初めにdf_train.csvを読み込んでPN値と株価の変化を表示することを行います。

# pn.rates_diff, exchange_rates_diffを表示して上記のプログラムの概要を把握してください。
rates_fd = open('./6050_stock_price_prediction_data/df_train.csv', 'r')
rates_fd.readline()  #1行ごとにファイル終端まで全て読み込んでいます。
next(rates_fd)  # 先頭の行を飛ばしています。

exchange_dates = []

pn_rates = []
pn_rates_diff = []

exchange_rates = []
exchange_rates_diff = []

prev_pn = df_train['pn'][0]
prev_exch = df_train['終値'][0]

for line in rates_fd:
    splited = line.split(",")
    time = splited[0]   # table.csvの１列目日付
    pn_val = float(splited[1])   # table.csvの２列目PN値
    exch_val = float(splited[2])  # table.csvの３列目株価の終値
    exchange_dates.append(time)  # 日付

    pn_rates.append(pn_val)
    pn_rates_diff.append(pn_val - prev_pn)   # PN値の変化

    exchange_rates.append(exch_val)
    exchange_rates_diff.append(exch_val - prev_exch)   # 株価の変化

    prev_pn = pn_val
    prev_exch = exch_val
rates_fd.close()
print(pn_rates_diff)
print(exchange_rates_diff)

株価予測その3

# 3日間ごとのPN値と株価の変化を表示してみます。

import numpy as np
INPUT_LEN = 3
data_len = len(pn_rates_diff)
tr_input_mat = []
tr_angle_mat = []

for i in range(INPUT_LEN, data_len):
    tmp_arr = []
    for j in range(INPUT_LEN):
        tmp_arr.append(exchange_rates_diff[i-INPUT_LEN+j])
        tmp_arr.append(pn_rates_diff[i-INPUT_LEN+j])   
    tr_input_mat.append(tmp_arr)  # i日目の直近3日間の株価とネガポジの変化

    if exchange_rates_diff[i] >= 0:  # i日目の株価の上下、プラスなら1、マイナスなら0
        tr_angle_mat.append(1)
    else:
        tr_angle_mat.append(0)   
train_feature_arr = np.array(tr_input_mat)
train_label_arr = np.array(tr_angle_mat)

# train_feature_arr, train_label_arrを表示して上のコードの概要を把握してください。
print(train_feature_arr)
print(train_label_arr)

株価の予測(まとめ)

# test_feature_arr, test_label_arrを同様に作成してください。
rates_fd = open('./6050_stock_price_prediction_data/df_test.csv', 'r')
rates_fd.readline()  #1行ごとにファイル終端まで全て読み込んでいます。
next(rates_fd)  # 先頭の行を飛ばしています。

exchange_dates = []

pn_rates = []
pn_rates_diff = []

exchange_rates = []
exchange_rates_diff = []

prev_pn = df_test['pn'][0]
prev_exch = df_test['終値'][0]

for line in rates_fd:
    splited = line.split(",")
    time = splited[0]   # table.csvの１列目日付
    pn_val = float(splited[1])   # table.csvの２列目PN値
    exch_val = float(splited[2])  # table.csvの３列目株価の終値
    exchange_dates.append(time)  # 日付

    pn_rates.append(pn_val)
    pn_rates_diff.append(pn_val - prev_pn)   # PN値の変化

    exchange_rates.append(exch_val)
    exchange_rates_diff.append(exch_val - prev_exch)   # 株価の変化

    prev_pn = pn_val
    prev_exch = exch_val
rates_fd.close()
INPUT_LEN = 3
data_len = len(pn_rates_diff)
test_input_mat = []
test_angle_mat = []

for i in range(INPUT_LEN, data_len):
    test_arr = []
    for j in range(INPUT_LEN):
        test_arr.append(exchange_rates_diff[i - INPUT_LEN + j])
        test_arr.append(pn_rates_diff[i - INPUT_LEN + j])   
    test_input_mat.append(test_arr)  # i日目の直近3日間の株価とネガポジの変化

    if exchange_rates_diff[i] >= 0:  # i日目の株価の上下、プラスなら1、マイナスなら0
        test_angle_mat.append(1)
    else:
        test_angle_mat.append(0)   
test_feature_arr = np.array(test_input_mat)
test_label_arr = np.array(test_angle_mat)

# train_feature_arr, train_label_arr,test_feature_arr, test_label_arrを特徴量にして、予測モデル(ロジスティック回帰、SVM、ランダムフォレスト)を構築し予測精度を計測してください。
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC

for model in [LogisticRegression(), RandomForestClassifier(n_estimators=200, max_depth=8, random_state=0), SVC()]:
    model.fit(train_feature_arr, train_label_arr)
    print("--Method:", model.__class__.__name__, "--")
    print("Cross validatin scores:{}".format(model.score(test_feature_arr, test_label_arr)))

日経平均株価の時系列データの取得

時系列データの取得その1

日経平均株価の過去のcsvデータを取ってきて保存します。

# csvを読み込む時のコード
import pandas as pd
from io import StringIO
import urllib

def read_csv(url):
    res = urllib.request.urlopen(url)
    res=res.read().decode('shift_jis')
    df = pd.read_csv(StringIO( res) )
    return df

# 日経平均株価の時系列データの取得
import pandas as pd
from io import StringIO
import urllib

# 上記の関数を使い日経平均株価の時系列データを取得
url = "https://indexes.nikkei.co.jp/nkave/historical/nikkei_stock_average_daily_jp.csv"
def read_csv(url):
    res = urllib.request.urlopen(url)
    res = res.read().decode('shift_jis')
    df = pd.read_csv(StringIO(res))
    # 必要のない最後の行を取り除いています
    df = df.drop(df.shape[0]-1)
    return df

# dfというdataframeで保存し、出力
df = read_csv(url)
# 出力
df

時系列データの取得その2

時系列データの終値に注目して予測を行います。

# 時系列データの取得その1,日経平均株価の時系列データの取得コードを載せます。
# indexを日付にした後、時系列にします
df["データ日付"] = pd.to_datetime(df["データ日付"], format='%Y/%m/%d')
df = df.set_index('データ日付')

# カラムから'始値', '高値', '安値'を取り除いて、日付が古い順に並べます
df = df.drop(['始値', '高値', '安値'], axis=1)
df = df.sort_index(ascending=True)
df

時系列データの取得その3

import pandas as pd

# dfとdf_tweetsの二つのテーブルをindexをキーとして結合し、Nanを消去
df_tweets = pd.read_csv('./6050_stock_price_prediction_data/df_tweets.csv', index_col='date')
table = df_tweets.join(df, how='right').dropna()
# table.csvとして出力
table.to_csv("./6050_stock_price_prediction_data/table.csv")
table

株価予測

株価予測その1

Python: 株価予測その①で説明したように
今回はテクニカル分析を用いて株価の予測を行います。

過去三日間の日経平均株価の時系列の変化とPN値の変化を特徴量にして
次の日の株価の上下の予測を行います。

訓練データとテストデータの二つに分け、訓練データを標準化したのち、訓練データの平均と分散を用いてテストデータの標準化を行います。

# 基本雛形
from sklearn.model_selection import train_test_split

X = table.values[:, 0]
y = table.values[:, 1]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0, shuffle=False)
X_train_std = (X_train - X_train.mean()) / X_train.std()
X_test_std = (X_test - X_train.mean()) / X_train.std()

# 学習データの作成
from sklearn.model_selection import train_test_split

X = table.values[:, 0]
y = table.values[:, 1]

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=0, shuffle=False)
X_train_std = (X_train - X_train.mean()) / X_train.std()
X_test_std = (X_test - X_train.mean()) / X_train.std()

# df_trainというテーブルを作りそこにindexを日付、カラム名をpn値、終値にしてdf_train.csvという名前でdataフォルダ内に出力
df_train = pd.DataFrame(
    {'pn': X_train_std,
     '終値': y_train},
    columns=['pn', '終値'],
    index=table.index[:len(X_train_std)])
df_train.to_csv('./6050_stock_price_prediction_data/df_train.csv')

# テストデータについても同様にdf_testというテーブルを作り、df_test.csvという名前でdataフォルダ内に出力
df_test = pd.DataFrame(
    {'pn': X_test_std,
     '終値': y_test},
    columns=['pn', '終値'],
    index=table.index[len(X_train_std):])
df_test.to_csv('./6050_stock_price_prediction_data/df_test.csv')

株価予測その2

# まず初めにdf_train.csvを読み込んでPN値と株価の変化を表示することを行います。

# pn.rates_diff, exchange_rates_diffを表示して上記のプログラムの概要を把握してください。
rates_fd = open('./6050_stock_price_prediction_data/df_train.csv', 'r')
rates_fd.readline()  #1行ごとにファイル終端まで全て読み込んでいます。
next(rates_fd)  # 先頭の行を飛ばしています。

exchange_dates = []

pn_rates = []
pn_rates_diff = []

exchange_rates = []
exchange_rates_diff = []

prev_pn = df_train['pn'][0]
prev_exch = df_train['終値'][0]

for line in rates_fd:
    splited = line.split(",")
    time = splited[0]   # table.csvの１列目日付
    pn_val = float(splited[1])   # table.csvの２列目PN値
    exch_val = float(splited[2])  # table.csvの３列目株価の終値
    exchange_dates.append(time)  # 日付

    pn_rates.append(pn_val)
    pn_rates_diff.append(pn_val - prev_pn)   # PN値の変化

    exchange_rates.append(exch_val)
    exchange_rates_diff.append(exch_val - prev_exch)   # 株価の変化

    prev_pn = pn_val
    prev_exch = exch_val
rates_fd.close()
print(pn_rates_diff)
print(exchange_rates_diff)

株価予測その3

# 3日間ごとのPN値と株価の変化を表示してみます。

import numpy as np
INPUT_LEN = 3
data_len = len(pn_rates_diff)
tr_input_mat = []
tr_angle_mat = []

for i in range(INPUT_LEN, data_len):
    tmp_arr = []
    for j in range(INPUT_LEN):
        tmp_arr.append(exchange_rates_diff[i-INPUT_LEN+j])
        tmp_arr.append(pn_rates_diff[i-INPUT_LEN+j])   
    tr_input_mat.append(tmp_arr)  # i日目の直近3日間の株価とネガポジの変化

    if exchange_rates_diff[i] >= 0:  # i日目の株価の上下、プラスなら1、マイナスなら0
        tr_angle_mat.append(1)
    else:
        tr_angle_mat.append(0)   
train_feature_arr = np.array(tr_input_mat)
train_label_arr = np.array(tr_angle_mat)

# train_feature_arr, train_label_arrを表示して上のコードの概要を把握してください。
print(train_feature_arr)
print(train_label_arr)

株価の予測(まとめ)

# test_feature_arr, test_label_arrを同様に作成してください。
rates_fd = open('./6050_stock_price_prediction_data/df_test.csv', 'r')
rates_fd.readline()  #1行ごとにファイル終端まで全て読み込んでいます。
next(rates_fd)  # 先頭の行を飛ばしています。

exchange_dates = []

pn_rates = []
pn_rates_diff = []

exchange_rates = []
exchange_rates_diff = []

prev_pn = df_test['pn'][0]
prev_exch = df_test['終値'][0]

for line in rates_fd:
    splited = line.split(",")
    time = splited[0]   # table.csvの１列目日付
    pn_val = float(splited[1])   # table.csvの２列目PN値
    exch_val = float(splited[2])  # table.csvの３列目株価の終値
    exchange_dates.append(time)  # 日付

    pn_rates.append(pn_val)
    pn_rates_diff.append(pn_val - prev_pn)   # PN値の変化

    exchange_rates.append(exch_val)
    exchange_rates_diff.append(exch_val - prev_exch)   # 株価の変化

    prev_pn = pn_val
    prev_exch = exch_val
rates_fd.close()
INPUT_LEN = 3
data_len = len(pn_rates_diff)
test_input_mat = []
test_angle_mat = []

for i in range(INPUT_LEN, data_len):
    test_arr = []
    for j in range(INPUT_LEN):
        test_arr.append(exchange_rates_diff[i - INPUT_LEN + j])
        test_arr.append(pn_rates_diff[i - INPUT_LEN + j])   
    test_input_mat.append(test_arr)  # i日目の直近3日間の株価とネガポジの変化

    if exchange_rates_diff[i] >= 0:  # i日目の株価の上下、プラスなら1、マイナスなら0
        test_angle_mat.append(1)
    else:
        test_angle_mat.append(0)   
test_feature_arr = np.array(test_input_mat)
test_label_arr = np.array(test_angle_mat)

# train_feature_arr, train_label_arr,test_feature_arr, test_label_arrを特徴量にして、予測モデル(ロジスティック回帰、SVM、ランダムフォレスト)を構築し予測精度を計測
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.svm import SVC

for model in [LogisticRegression(), RandomForestClassifier(n_estimators=200, max_depth=8, random_state=0), SVC()]:
    model.fit(train_feature_arr, train_label_arr)
    print("--Method:", model.__class__.__name__, "--")
    print("Cross validatin scores:{}".format(model.score(test_feature_arr, test_label_arr)))

pythonで動くものを作りたい！と思い、『Pythonでつくるゲーム開発入門講座』(実践編含む)で勉強中です。

なぜpythonのtkinterなのか→私の最終目的はゲーム開発ではなく、スクレイピングなどのデータ収集や、データ分析、業務効率化など。よって、ゲーム開発に適したC++やUnity？でなく、pythonでゲーム作成。pygameライブラリを使わないのも同じ理由。

スクレイピングやデータ分析、機械学習をピンポイントで勉強しても、今のところ応用方法が思いつかないので、遠回りになるけれど、ゲームを作成しながら、プログラミングを覚えようという魂胆。NCPを動かすのに、機械学習とかディープラーニングを組み込んでそのあたりの実践経験も積めたらなぁとか欲張ってます。

ある程度学習出来たので、サンプルコードを参考（というか改造）にして、不思議なダンジョンゲームっぽいものを作ってみました。
風来の試練やトルネコを思い浮かべる方が多いと思いますが、私は不思議なダンジョンと言えば、チョコボの不思議なダンジョンを思い浮かべます。（あとはテリーのワンダーランド？）最終的にはチョコボに近い'何か'を作りたいでうすね。
とりあえず、コードとキャラチップをgithubに保存してみました。
キャラチップはぴぽや倉庫さんより拝借いたしました。 マップを自動生成する、maze_maker.pyとゲーム本体のtkmaze.pyに分けてみました。

maze_maker.py

import random

class maze_maker:
    """
    ダンジョンを自動生成する
    """

    def __init__(self,MAZE_W,MAZE_H):
        self.MAZE_W = MAZE_W
        self.MAZE_H = MAZE_H
        self.maze = [[0]*self.MAZE_W for y in range(self.MAZE_H)]
        self.DUNGEON_W = MAZE_W*3
        self.DUNGEON_H = MAZE_H*3
        self.dungeon = [[0]*self.DUNGEON_W for y in range(self.DUNGEON_H)]

    def make_maze(self):
        """
        迷路を作る
        """
        XP = [ 0, 1, 0,-1]
        YP = [-1, 0, 1, 0]

        #周囲の柱
        for x in range(self.MAZE_W):
            self.maze[0][x] = 1
            self.maze[self.MAZE_H-1][x] = 1
        for y in range(self.MAZE_H):
            self.maze[y][0] = 1
            self.maze[y][self.MAZE_W-1] = 1

        #中を空っぽに
        for y in range(1,self.MAZE_H-1):
            for x in range(1,self.MAZE_W-1):
                self.maze[y][x] = 0

        #柱
        for y in range(2,self.MAZE_H-2,2):
            for x in range(2,self.MAZE_W-2,2):
                self.maze[y][x] = 1

        for y in range(2,self.MAZE_H-2,2):
            for x in range(2,self.MAZE_W-2,2):
                d = random.randint(0,3)
                if x > 2:
                    d = random.randint(0,2)
                self.maze[y+YP[d]][x+XP[d]] = 1

    def make_dungeon(self):
        """
        迷路からダンジョンを作る
        """
        self.make_maze()
        for y in range(self.DUNGEON_H):
            for x in range(self.DUNGEON_W):
                self.dungeon[y][x] = 9
        for y in range(1,self.MAZE_H-1):
            for x in range(1,self.MAZE_W-1):
                dx = x*3+1
                dy = y*3+1
                if self.maze[y][x] == 0:
                    if random.randint(0,99) < 20:
                        for ry in range(-1,2):
                            for rx in range(-1,2):
                                self.dungeon[dy+ry][dx+rx] = 0
                    else:
                        self.dungeon[dy][dx] = 0
                        if self.maze[y-1][x] == 0:
                            self.dungeon[dy-1][dx] = 0
                        if self.maze[y+1][x] == 0:
                            self.dungeon[dy+1][dx] = 0
                        if self.maze[y][x-1] == 0:
                            self.dungeon[dy][dx-1] = 0
                        if self.maze[y][x+1] == 0:
                            self.dungeon[dy][dx+1] = 0


    def put_event(self):
        while True:
            x = random.randint(3,self.DUNGEON_W-4)
            y = random.randint(3,self.DUNGEON_H-4)
            if(self.dungeon[y][x] == 0):
                for ry in range(-1,2):
                    for rx in range(-1,2):
                        self.dungeon[y+ry][x+rx] = 0
                self.dungeon[y][x] = 1
                break
        for i in range(60):
            x = random.randint(3,self.DUNGEON_W-4)
            y = random.randint(3,self.DUNGEON_H-4)
            if(self.dungeon[y][x] == 0):
                self.dungeon[y][x] = random.choice([2,3,3,3,4])

tkmaze.py

"""
冥土で冥土探索。それは終わりなき旅。ゴールは無くて、敵を避けつつひたすら階段を降りていく。
1歩ごとにスコアカウント。階段を降りるとプラス。ハイスコアを目指そう！
"""
import tkinter
import maze_maker
from PIL import Image,ImageTk
import random

#キー入力
key = ''
koff = False
def key_down(e):
    global key,koff
    key = e.keysym
    koff = False

def key_up(e):
    global koff
    koff = True

CHIP_SIZE = 32
DIR_UP = 3
DIR_DOWN = 0
DIR_LEFT = 1
DIR_RIGHT = 2
chara_x = chara_y = 146
chara_d = chara_a = 0
obj_a = 0
emy_num = 4
emy_list_x = [0]*emy_num
emy_list_y = [0]*emy_num
emy_list_d = [0]*emy_num
emy_list_a = [0]*emy_num
ANIMATION = [1,0,1,2]
#素材は「ぴぽや http://blog.pipoya.net/」様より
imgplayer_pass = 'image/charachip01.png'
emy_img_pass = 'image/pipo-charachip019.png'
emy2_img_pass = 'image/hone.png'
emy3_img_pass = 'image/majo.png'
emy3_kageimg_pass = 'image/majo_kage.png'
takara_img_pass = 'image/pipoya_mcset1_obj01.png'
obj_pass = 'image/pipoya_mcset1_obj02.png'
obj2_pass = 'image/pipo-hikarimono005.png'
yuka_pass = 'image/pipoya_mcset1_at_gravel1.png'
kebe_pass = 'image/pipoya_mcset1_bridge01.png'

map_data = maze_maker.maze_maker(11,7)
map_data.make_dungeon()
map_data.put_event()

tmr = 0
idx = 1
floor_count = 0
item_count = 0
pl_life=150
pl_stamina = 150
pl_damage = 0

def move_player():
    global chara_x,chara_y,chara_a,chara_d,pl_stamina,pl_life
    if key == 'Up':
        chara_d = DIR_UP
        check_wall()
    if key == 'Down':
        chara_d = DIR_DOWN
        check_wall()
    if key == 'Left':
        chara_d = DIR_LEFT
        check_wall()
    if key == 'Right':
        chara_d = DIR_RIGHT
        check_wall()
    check_event()
    if tmr%4 == 0:
        if pl_stamina > 0:
            pl_stamina -= 1
        else:
            pl_life -= 1
            if pl_life <= 0:
                pl_life = 0
                idx = 2
    chara_a = chara_d*3 + ANIMATION[tmr%4]


def emy_set(emy_num):
    while True:
        x = random.randint(3,map_data.DUNGEON_W-4)
        y = random.randint(3,map_data.DUNGEON_H-4)
        if map_data.dungeon[y][x] == 0:
            emy_list_x[emy_num] = x*CHIP_SIZE+(CHIP_SIZE//2)
            emy_list_y[emy_num] = y*CHIP_SIZE+(CHIP_SIZE//2)
            break


def draw_text(txt):
    st_fnt = ('Times New Roman',60)
    canvas.create_text(200,200,text=txt,font=st_fnt,fill='red',tag='SCREEN')


def damage_cal(pl_damage):
    global pl_life,idx
    if pl_life <= pl_damage:
        pl_life = 0
        idx = 2
    else:
        pl_life -= pl_damage


def move_emy(emy_num):
    cy = int(emy_list_y[emy_num]//CHIP_SIZE)
    cx = int(emy_list_x[emy_num]//CHIP_SIZE)
    emy_list_d[emy_num] = random.randint(0,3)
    if emy_list_d[emy_num] == DIR_UP:
        if map_data.dungeon[cy-1][cx] != 9:
            emy_list_y[emy_num] -= 32
    if emy_list_d[emy_num] == DIR_DOWN:
        if map_data.dungeon[cy+1][cx] != 9:
            emy_list_y[emy_num] += 32
    if emy_list_d[emy_num] == DIR_LEFT:
        if map_data.dungeon[cy][cx-1] != 9:
            emy_list_x[emy_num] -= 32
    if emy_list_d[emy_num] == DIR_RIGHT:
        if map_data.dungeon[cy][cx+1] != 9:
            emy_list_x[emy_num] += 32
    emy_list_a[emy_num] = emy_list_d[emy_num]*3 + ANIMATION[tmr%4]
    if abs(emy_list_x[emy_num]-chara_x) <= 30 and abs(emy_list_y[emy_num]-chara_y) <= 30:
        pl_damage = 10*random.choice([1,2,2,3,3])
        damage_cal(pl_damage)


def obj_animation():
    global obj_a
    obj_a = ANIMATION[tmr%4]


def check_wall():
    global chara_x,chara_y,chara_a,chara_d
    cy = int(chara_y//CHIP_SIZE)
    cx = int(chara_x//CHIP_SIZE)
    if chara_d == DIR_UP:
        if map_data.dungeon[cy-1][cx] != 9:
            chara_y -= 32
    if chara_d == DIR_DOWN:
        if map_data.dungeon[cy+1][cx] != 9:
            chara_y += 32
    if chara_d == DIR_LEFT:
        if map_data.dungeon[cy][cx-1] != 9:
            chara_x -= 32
    if chara_d == DIR_RIGHT:
        if map_data.dungeon[cy][cx+1] != 9:
            chara_x += 32

def check_event():
    global chara_x,chara_y,chara_a,chara_d,idx,pl_damage
    global floor_count,emy_count,item_count,pl_life,pl_stamina
    cy = int(chara_y//CHIP_SIZE)
    cx = int(chara_x//CHIP_SIZE)
    if map_data.dungeon[cy][cx] == 1:
        #ワープの魔法陣にのった
        map_data.make_dungeon()
        map_data.put_event()
        emy_set(0)
        emy_set(1)
        emy_set(2)
        floor_count += 1
        chara_x = chara_y = 146
    if map_data.dungeon[cy][cx] == 2:
        #トラップの魔法陣に乗った
        if item_count > 0:
            item_count -= 1
            map_data.dungeon[cy][cx] = 0
        else:
            pl_damage = 5*random.choice([1,2,2,3,4,3])
            damage_cal(pl_damage)
            map_data.dungeon[cy][cx] = 0
    if map_data.dungeon[cy][cx] == 3:
        #アイテムに接触
        item_count += 1
        map_data.dungeon[cy][cx] = 0
    if map_data.dungeon[cy][cx] == 4:
        #食料に接触
        pl_stamina_recover = 5*random.choice([1,2,2,3,4,3])
        if pl_stamina + pl_stamina_recover > 150:
            pl_stamina = 150
        else:
            pl_stamina += pl_stamina_recover
        map_data.dungeon[cy][cx] = 0


def split_chip(chip_pass,chip_img_x,chip_img_y):
    '''
    複数のチップを１単位のチップに分割する
    '''
    chip_list = []
    for cy in range(0,chip_img_y,CHIP_SIZE):
        for cx in range(0,chip_img_x,CHIP_SIZE):
            chip_list.append(ImageTk.PhotoImage(Image.open(chip_pass).crop((cx,cy,cx+CHIP_SIZE,cy+CHIP_SIZE))))
    return chip_list


def draw_screen():
    st_fnt = ('Times New Roman',30)
    canvas.delete('SCREEN')
    for my in range(len(map_data.dungeon)):
        for mx in range(len(map_data.dungeon[0])):
            if map_data.dungeon[my][mx] != 9:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=yuka_img[8],tag='SCREEN')
            if map_data.dungeon[my][mx] == 1:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=obj2_img[obj_a],tag='SCREEN')
            if map_data.dungeon[my][mx] == 2:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=obj2_img[6+obj_a],tag='SCREEN')
            if map_data.dungeon[my][mx] == 3:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=takara_img[5],tag='SCREEN')
            if map_data.dungeon[my][mx] == 4:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=obj_img[25],tag='SCREEN')
            if map_data.dungeon[my][mx] == 9:
                canvas.create_image(mx*CHIP_SIZE+(CHIP_SIZE//2),my*CHIP_SIZE+(CHIP_SIZE//2),image=kabe_img[28],tag='SCREEN')
    canvas.create_image(chara_x,chara_y,image=imgplayer[chara_a],tag='SCREEN')
    canvas.create_image(emy_list_x[0],emy_list_y[0],image=emy_img[emy_list_a[0]],tag='SCREEN')
    canvas.create_image(emy_list_x[1],emy_list_y[1],image=emy2_img[emy_list_a[1]],tag='SCREEN')
    canvas.create_image(emy_list_x[2],emy_list_y[2],image=emy3_img[emy_list_a[2]],tag='SCREEN')
    canvas.create_image(emy_list_x[2],emy_list_y[2],image=emy3_kageimg[emy_list_a[2]],tag='SCREEN')
    canvas.create_text(1110,50,text='{} 階'.format(floor_count),font=st_fnt,fill='black',tag='SCREEN')
    canvas.create_text(1110,100,text='{} 個'.format(item_count),font=st_fnt,fill='black',tag='SCREEN')
    canvas.create_rectangle(1060,135,1210,160,fill='black',tag='SCREEN')
    canvas.create_rectangle(1060,135,1060+pl_life,160,fill='limegreen',tag='SCREEN')
    canvas.create_text(1130,148,text='LIFE',font=('Times New Roman',15),fill='white',tag='SCREEN')
    canvas.create_rectangle(1060,165,1210,190,fill='black',tag='SCREEN')
    canvas.create_rectangle(1060,165,1060+pl_stamina,190,fill='blue',tag='SCREEN')
    canvas.create_text(1130,178,text='STAMINA',font=('Times New Roman',15),fill='white',tag='SCREEN')
    canvas.create_text(1155,210,text='{}のダメージを受けた！'.format(pl_damage),font=('Times New Roman',15),fill='black',tag='SCREEN')


def main():
    global tmr,koff,key,idx
    tmr += 1
    draw_screen()
    if idx == 1:
        if tmr == 1:
            for emy in range(0,emy_num):
                emy_set(emy)
        move_player()
        obj_animation()
        if tmr%2 == 0:
            for emy in range(0,emy_num):
                move_emy(emy)
        if pl_life == 0:
            idx = 2
    if idx == 2:
        draw_text('You Died')
        if tmr == 20:
            idx = 1
            print('hiu')

    if koff == True:
        key = ''
        koff = False

    root.after(130,main)



root = tkinter.Tk()
root.title('メイドで冥土探検！')
root.bind('<KeyPress>',key_down)
root.bind('<KeyRelease>',key_up)
canvas = tkinter.Canvas(width=1256,height=864)
imgplayer = split_chip(imgplayer_pass,96,128)
emy_img = split_chip(emy_img_pass,96,128)
emy2_img = split_chip(emy2_img_pass,96,128)
emy3_img = split_chip(emy3_img_pass,96,128)
emy3_kageimg = split_chip(emy3_kageimg_pass,96,128)
takara_img = split_chip(takara_img_pass,256,64)
obj_img = split_chip(obj_pass,256,224)
obj2_img = split_chip(obj2_pass,96,128)
yuka_img = split_chip(yuka_pass,64,160)
kabe_img = split_chip(kebe_pass,256,192)
canvas.pack()
main()
root.mainloop()

プレイ画面。

全く面白く無いですが、マップの生成、プレイヤーが足踏みしながら方向を変える、ワープゾーンがアニメーション、敵キャラの配置・移動、ダメージ計算・・・など、とりあえずダンジョンゲームに必要な基礎的な処理はまずまず揃ったのでは？と思います。ちなみに、敵への攻撃はまだ実装できていません（汗

改良したい事

• 敵がバカで適当にうろうろしているだけ。追いかけてくるとか、移動スピードを変えるとか組み込んでみたい
• 攻撃できるようにして、敵を撃破できるようにする
• (かなり発展？)プレイヤーのレベルやらステータスとか作り込む
• 【願望】敵に強化学習or機械学習を組み込む

コードもまだまだ整理出来ていない気がするので、少しずつ改良したいです。

こだわり？

ピクセル数を指定してキャラチップを分解する関数。色々情報を集めながら頑張って考えました。

はじめに

数理最適化問題（数理計画問題）の中で、線型最適化問題（線型計画問題）や整数最適化問題（整数計画問題）（※ただし数式がすべて線型のもの）を解いてくれるPythonパッケージとして、PuLPがあります。

• 参考Qiita記事： Python+PuLPによるタダで仕事に使える数理最適化

PuLPは、解きたい対象をPython上で数式として記述できるモデリングインターフェース部分と、数理最適化ソルバーと呼ばれる、式として記述したものを解いてくれる計算エンジン部分とが、きれいに分かれています。PuLPには、数理最適化ソルバーとして、 無償で商用利用可能な COIN-CBC（ 公式サイト（GitHub） ）の、とあるバージョンが同こんされているほか、自分で入手したほかの数理最適化ソルバーを呼び出すこともできます。ということは、COIN-CBCより高性能な数理最適化ソルバーを用意すれば、より短い計算時間で解くことができるのですが、無償で商用利用可能なソルバーの中でCOIN-CBCより速いものはMIPCLくらいしかなく、そのMIPCLも、2020年5月現在、入手性に謎の部分がでてきました。

• 参考Qiita記事： ついに使い物になるフリーの数理最適化ソルバーが登場？　MIPCL

ところで、整数最適化問題を解くアルゴリズムは、特にその中の分枝限定法と呼ばれる部分について、並列処理が可能であり、それにより計算時間をけっこう短縮できます。有償のソルバーやMIPCL、そして公式サイトで配布されているCOIN-CBCは、マルチスレッドで動作するのですが、PuLPに同こんされているCOIN-CBCは、少なくともWindows版については、マルチスレッドで動作しないのです。

そこで、PuLPから呼び出す数理最適化ソルバーを、マルチスレッド対応版のCOIN-CBCに変更することで、計算時間を短縮する方法を紹介します。今回は、速報版として、お手軽にできる方法を扱います。ついでに、計算時間の上限を指定する方法にも触れます。

※ Macの方は、自分でインストールした CBCソルバをPuLPで使用する の記事が参考になると思います。

実験環境

• Windows 10 1909 64bit
  • 管理者権限を持つアカウントで作業
    • （作業中に管理者権限を求められた記憶はなし）
• Python 3.7 64bit (Anaconda3 2020.02 64bit)
  • ユーザー領域にインストール（Anacondaインストール時に「Just Me」を選択）
• PuLP 2.1
• 2020/05/18にmasterブランチからビルドされたCOIN-CBC
• Visual Studio Code 1.45.1
  • ユーザー領域にインストール

ダウンロード

Python+PuLPによるタダで仕事に使える数理最適化 の記事などに従って、PuLPのダウンロードとインストール、軽い動作確認はできているものとします（最適化の計算では、特に指定しない場合、PuLP同こん版のCOIN-CBCが使用されます）。

今回、追加作業として、 https://bintray.com/coin-or/download/Cbc/master#files の Cbc-master-x86_64-w64-mingw32.zip をダウンロードします。ダウンロードしたら、zipファイルなので、展開します。

• このバイナリーは、masterブランチからビルドされたもののようです。上記のサイトの中や COIN-CBCの公式サイト（GitHub） を探していくと、バージョン番号のついたバイナリーも見つかりますが、そちらはライブラリーの依存関係を自分で解決する必要があるようです。
  • masterブランチからビルドされたものは、バージョンが固定された安定版とは限らないので、使用することに不安はありますね…自己責任の度合いがより強くなります。
• Cbc-master-win64-msvc16-mt.zip （Visual Studio 2019でコンパイルしたもの？）などほかのバイナリーは、なぜかマルチスレッドに対応していませんでした。

コード

ダウンロードして展開したCOIN-CBCが、 C:\Users\(ユーザー名)\Desktop\Cbc-master-x86_64-w64-mingw32\bin\cbc.exe にあると想定します。

こちらの記事の例(2) の コード pulp_problem_2.py を取り上げて説明しますと、

pulp_problem_2.py

# （前略）
# 計算
# ソルバー指定
solver = pulp.PULP_CBC_CMD()
# （後略）

の部分を、

pulp_problem_2_mt.py

# ソルバー指定
solver = pulp.COIN_CMD(path=r'C:\Users\(ユーザー名)\Desktop\Cbc-master-x86_64-w64-mingw32\bin\cbc.exe', threads=8, maxSeconds=120.5)

に変えれば、マルチスレッドで動作します。

• path=(COIN-CBCのバイナリーのパス) で、呼び出すCOIN-CBCのバイナリーのあるところを指定します。
  • 先頭に r のついた文字列は raw文字列 というやつで、Windowsのファイルやフォルダーのパスを指定する際に使うと便利なものです。
• threads=(スレッド数) で、COIN-CBCで使用するスレッド数を指定します。いくつがよいかは、下記のベンチマーク結果が参考になると思います。
• 今回の記事の主題から外れますが、 maxSeconds=(計算時間上限（秒）) で、計算時間の上限を指定できます。
  • 指定時間内に計算が終わる場合もあります。その場合、（最適解が存在するインスタンスであれば）最適解が返ってきて、かつ、その解が最適解であることが計算を通して理論的に保証されています。
    • （インスタンスが非有界（目的関数を無限に小さく（大きく）できる）や実行不可能（制約を満たす解が存在しない）である場合は、そうであることが指定時間内に判明したということになります。）
  • 計算が終わらないまま指定時間に達した場合は、それまでに見つかった中でいちばんよい解が返ってきます。
    • 返ってきた解は、実は最適解であるかもしれないし、そうでないかもしれません。仮に最適解であったとしても、最適解であることは理論的に保証されていません。
    • この「指定時間内で見つかった解を返す」機能の使い道のひとつとして、問題に対して自前で（メタ）ヒューリスティクスのコードを組んで指定時間内でそこそこよい解を探す代わりに、数理最適化ソルバーに投げて同じ時間内で見つかったそこそこよい解を返してもらうことが考えられます。
    • 指定時間内に解が1つも見つからない可能性もあります。その場合、そもそも実行可能解（制約を満たす解）が存在しないのかもしれないし、存在はするものの数理最適化ソルバーが見つけられていないのかもしれません。

ベンチマーク

前の記事 と同じく、グラフクラスタリング（与えられた無向グラフを、いくつかの密な部分グラフに分割する問題）に関する こちらの論文 の定式化を解いてみました。環境はIntel Core i9-9900K（8コア16スレッド）です。
この論文の定式化に基いてインスタンスを解いた印象としては、最適解を見つけるのにも少し時間がかかるものの、それ以上に、見つかった解が最適であることを保証するために多くの時間がかかるようなタイプであるということが挙げられます。

インスタンス	PuLP 2.1付属のCOIN-CBC (1スレ)	2020/05/18版COIN-CBC (1スレ)	2020/05/18版COIN-CBC (8スレ)	2020/05/18版COIN-CBC (16スレ)
1	21.7	22.7	6.9	13.4
2	10.1	11.2	4.3	5.6
3	28.0	20.3	7.2	10.3
4	9.8	19.8	5.4	7.4

（単位：秒）
（おおむね、インスタンスの数字が大きいほど、インスタンスのサイズが大きい）

まず、1スレッドで動作するPuLP 2.1付属のCOIN-CBCのバイナリーと、今回入手したCOIN-CBCのバイナリーを1スレッドで動作させた場合を比較すると、おおむね前者のほうが短時間でインスタンスを解いています。前者のバイナリーは、今回の実験環境の場合、 C:\Users\(ユーザー名)\Anaconda3\Lib\site-packages\pulp\solverdir\cbc\win\64\cbc.exe にあるのですが、これをコンソールから実行させたら、 Version: 2.9.0 Build Date: Feb 12 2015 と表示されました。2020/05/18版は、バージョン2.10.5から2か月ちょっとたったもののようです。数理最適化ソルバーでは、バージョンが上がると、中のチューニングの変更の結果として、かえって解く時間が長くなるインスタンスがでてくることはままあります。
今回入手したCOIN-CBCのバイナリーを、CPUの物理コア数である8スレッドで動作させると、1スレッドのときと比べて、最低でも40%弱の時間でインスタンスが解けました。コア数が1:8なので計算時間が12.5%まで短縮とはいきませんでしたが、なかなかよい結果です。
物理コア数を超えて論理コア数である16スレッドで動作させた場合は、8スレッドで動作させた場合に比べて時間がかかってしまいました。整数最適化は比較的重い処理であることから、SMT機能により別の論理コアとして見えているが物理的には同一のコアの中で、演算器の奪い合いが起こってパフォーマンスが下がったものと推測されます。こういった現象は、例えば行列演算ベンチマークとして知られているLINPACKでも起こります。最近の有償ソルバーだと、自動でCPUのSMT周りの特性をチェックして、スレッド生成数を自動調整してくれるようなのですが、今回入手したCOIN-CBCにはそういった機能はないようなので、自分で明示的にスレッド数 = 物理コア数と設定するのがよいようです。なお、各スレッドの処理をどの論理コアへ割りふるかについては、Windowsの場合はVistaか7のころから、OS側で別々の物理コアにいい感じに割りふってくれているようです。

まとめ

• PuLPにデフォルトでついてくる数理最適化ソルバーは1スレッドでしか動かないので、マルチスレッド動作版をダウンロードし、CPUの物理コア数ぶんのスレッドを指定して呼び出すと、問題を速く解けます。
• PuLPでは計算時間の上限を指定することができます。指定時間内に計算が終わる場合もあります。計算が終わらないまま指定時間に達した場合は、それまでに見つかった中でいちばんよい解が返ってきます。

これからやりたいこと

• マルチスレッド動作版をソースからコンパイルしたい。
• cbc.exe を呼び出すのではなく、（ソールからコンパイルして生成した）DLLを呼び出したい。
  • cbc.exe を呼び出す方式の場合、 PuLPがインスタンスを.mpsと呼ばれる形式のファイルに出力 → cbc.exe が.mpsファイルを読んで計算 → cbc.exe が解を.solと呼ばれる形式のファイルに出力 → PuLPが.solファイルを読み解をセット 、という処理の流れになっているが、規模の大きな問題の場合、ファイルの入出力に時間がかかる。
  • PuLPにはデフォルトで CoinMP.dll が付属しているが、32bit版であり、64bit環境では動かない模様？
• Google OR-Tools に付属しているCOIN-CBCはマルチスレッドで動作するのか否かを調査したい。

概要

海外ではエンジニアの面接においてコーディングテストというものが行われるらしく、多くの場合、特定の関数やクラスをお題に沿って実装するという物がメインである。

その対策としてLeetCodeなるサイトで対策を行うようだ。

早い話が本場でも行われているようなコーディングテストに耐えうるようなアルゴリズム力を鍛えるサイト。

せっかくだし人並みのアルゴリズム力くらいは持っておいた方がいいだろうということで不定期に問題を解いてその時に考えたやり方をメモ的に書いていこうかと思います。

Leetcode

ゼロから始めるLeetCode 目次

前回
ゼロから始めるLeetCode Day34「118. Pascal's Triangle」

基本的にeasyのacceptanceが高い順から解いていこうかと思います。

Twitterやってます。

問題

160. Intersection of Two Linked Lists

Top 100 Liked QuestionsのEasy問題の最後から二つ目の問題です。

問題としては、2つの単方向連結リストの共通部分が始まるノードを見つけてくださいというものです。

例が図で解説されており、諸事情でこちらに直接載せることはできないので各々が確認して頂けると幸いです。

解法

最初こう書いたら、

# Definition for singly-linked list.
# class ListNode:
#     def __init__(self, x):
#         self.val = x
#         self.next = None

class Solution:
    def getIntersectionNode(self, headA: ListNode, headB: ListNode) -> ListNode:
        if headA == None or headB == None:
            return None

        LNA = headA
        LNB = headB
        while LNA != LNB:
            if LNA == None:
                LNA = headB
            else:
                LNA.next

            if LNB == None:
                LNB == headA
            else:
                LNB.next
        return LNA
# Time Limit Exceeded

時間切れになったので、以下のようにwhile以降を内包表記で書き直してみたら行けました。

# Definition for singly-linked list.
# class ListNode:
#     def __init__(self, x):
#         self.val = x
#         self.next = None

class Solution:
    def getIntersectionNode(self, headA: ListNode, headB: ListNode) -> ListNode:
        if headA == None or headB == None:
            return None

        LNA = headA
        LNB = headB
        while LNA != LNB:
            LNA = headB if LNA == None else LNA.next
            LNB = headA if LNB == None else LNB.next
        return LNA
# Runtime: 168 ms, faster than 75.08% of Python3 online submissions for Intersection of Two Linked Lists.
# Memory Usage: 29.1 MB, less than 100.00% of Python3 online submissions for Intersection of Two Linked Lists.

なお、この問題の公式の解説の解き方ではBruteForce,HashMap,Two Pointerのいずれかを使うと良いよ！ってなってます。
気になるかたはそちらをチェックすることもお勧めします。

こっちの書き方の方が良いよ！とかこの言語で書いてみたよ！とかがあれば是非コメントしてみてください。

１　メトリクス・ログを取得できるのか？
pod cpu
pod memory
node disk

• advance container cpu container memory container filesystem

log

2 もれなく取得できているか？

＊メトリクスについて
前提
　監視対象：EC2
　監視項目：CPU使用率の平均値
　get_metric_statisticsのパラメータのstart-timeはCloudwatchがリクエストを受け取った時に分以下を切り捨てる仕様
　　ex)
　　start_timeが21:17:32、end_timeが21:18:32の場合、
　　21:17:00として切り捨てられる。レスポンスの内容は、21:17:00から21:17:32のデータポイントの値を返す

・リアルタイムの監視をした場合
get_metric_statisticsを実行したときの設定値
　Cloudwatchのメトリクス取得期間：60s

start_time：現在時刻 ー 60s
end_time：現在時刻
period：60s

現在時刻が58秒の場合、取得可能
それ以外の場合は値を取得できなかった

・現在時刻から１分遅れて監視する場合
getメソッドを実行したときの設定値
　Cloudwatchのメトリクス取得期間：60s

start_time：現在時刻 ー 2m
end_time：現在時刻　ー　1m
period：60s

問題なく取得できる!

→現在時刻から１分遅れて監視をすれば確実に値を取得できる

*ログについて
前提
　監視対象：EC2
　監視ログ：nginxのアクセスログ
　get_log_eventsのパラメータのstartTimeやendTimeはエポックミリ秒で表記する必要がある。
startTimeは指定した時間を含んだ時間として値を返す
endTimeは指定した時間を含まない時間として値を返す
ex)
startTimeを19:22:54.213、endTimeを19:23:54.213の場合、
19:22:54.213 ~ 19:23:54.212の期間にcloudwatch logsにあるログがレスポンスとして返ってくる

・リアルタイムの監視をした場合
リアルタイム監視をした場合、監視対象のサーバから値をcloudwatch logに送りAPIとして取得できるようになるまでの期間分のログデータが取れなくなる
→そのため、少し遅れての監視が必要になる
　値を取得できるまでに10~20秒かかるような仕様となっていた。（cloudwatch logsで確認）

・現在時刻から30秒遅れて監視する場合
30秒遅れた監視をすればログを回収できたが、値をAPIから取得できるまでの期間がドキュメントにも記載がないため、もう少し間隔を開けた方がいいかもしてない

・監視期間について（startTime, endTime）
startTimeとendTimeの指定についてはミリ秒単位であるため、スクリプト上で毎回リアルタイムを作って（time.time()）実行するとどうしても取得できない部分が出てきてしまう。
これは、zabbixがスクリプトを実行する間隔がミリ秒単位で正確であれば問題なく全てのログを取得できるが、おそらくそこまで正確では無いのでは無いか？
→解決策として、前回実行した時間を書き込んだファイルを作りそこから、次回実行時に値をとって実行するようにすれば、漏れることなく全てのログを取得できる。

3 結論
メトリクス
　現在時刻から１分遅れて監視をすれば確実に値を取得できる（詳細モニタリングに設定している場合（コンテナの場合はcloudwatch agentが監視項目の値を取得する間隔が60秒の場合））

ログ
　以下の二つを気を付けることで漏れることなく監視が可能
　・現在時刻から少なくとも30秒遅れて監視すること
　・前回実行した時間をファイルに書き出し、次回実行時にはそこから値（前回実行した時間）を取り出しCloudwatchの情報をとってくること

aws_zabbix.py

#!/bin/env python

import boto3
import json
import argparse
import os
import socket
import struct
import time
import calendar
from datetime import datetime
from datetime import timedelta

class Metric:
    def __init__(self, name="", namespace="", unit="", dimensions=[]):
        self.name = name
        self.namespace = namespace
        self.unit = unit
        self.dimensions = dimensions

class AwsZabbix:
    '''
    dimentionsは'PodName'にしています(コンテナでの監視の場合は書き加える必要あり)
    '''

    def __init__(self, region, access_key, secret, identity, hostname, service, timerange_min,
                 zabbix_host, zabbix_port):
        self.zabbix_host = zabbix_host
        self.zabbix_port = zabbix_port
        self.identity = identity
        self.hostname = hostname
        self.service = service
        self.timerange_min = timerange_min
        self.id_dimentions = {
            'ec2':'InstanceId',
            'rds':'DBInstanceIdentifier',
            'elb':'LoadBalancerName',
            'ebs':'VolumeId',
            'billing': 'Currency',
            'eks': 'PodName',
        }
        self.client = boto3.client(
            'cloudwatch',
            region_name=region,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret
        )
        self.sum_stat_metrics = [
            {'namespace': 'AWS/ELB', 'metricname': 'RequestCount'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_Backend_2XX'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_Backend_3XX'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_Backend_4XX'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_Backend_5XX'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_ELB_4XX'},
            {'namespace': 'AWS/ELB', 'metricname': 'HTTPCode_ELB_5XX'},
            {'namespace': 'ContainerInsights', 'metricname': 'pod_number_of_container_restarts'},
        ]
        self.max_stat_metrics = [
            {'namespace': 'ContainerInsights', 'metricname': 'pod_cpu_reserved_capacity'},
            {'namespace': 'ContainerInsights', 'metricname': 'pod_memory_reserved_capacity'},
        ]

    def __get_metric_list(self):
        resp = self.client.list_metrics(
            Dimensions = [
                {
                    'Name': self.id_dimentions[self.service],
                    'Value': ('USD' if self.service == "billing" else self.identity)
                }
            ]
        )
        metric_list = []
        for data in resp["Metrics"]:
            metric = Metric(name=data["MetricName"], namespace=data["Namespace"], dimensions=data["Dimensions"])
            if self.service == "elb":
                for dimension in data["Dimensions"]:
                    if dimension["Name"] == "AvailabilityZone":
                       metric.name = data["MetricName"] + "." + dimension["Value"]
            metric_list.append(metric)
        return metric_list

    def __get_metric_stats(self, metric_name, metric_namespace, servicename, timerange_min, dimensions, stat_type="Average", period_sec=60):
        if self.service == "billing":
            dimensions = [
                {
                    'Name': self.id_dimentions[self.service],
                    'Value': 'USD'
                }
            ]
            if servicename != "billing":
                dimensions.insert(0,
                    {
                        'Name': 'ServiceName',
                        'Value': servicename
                    }
                )
        # 現在時刻よりも1分遅れて取得する
        end_time = datetime.utcnow() - timedelta(minutes=1)
        start_time = end_time - timedelta(minutes=timerange_min)
        stats = self.client.get_metric_statistics(
            Namespace=metric_namespace,
            MetricName=metric_name,
            Dimensions=dimensions,
            StartTime=start_time,
            EndTime=end_time,
            Period=period_sec,
            Statistics=[stat_type],
        )
        return stats

    def __set_unit(self, metric_list):
        ret_val = []
        for metric in metric_list:
            servicename = self.service
            if self.service == "billing":
                metric.unit = 'USD'
            else:
                stats = self.__get_metric_stats(metric.name, metric.namespace, servicename, self.timerange_min)
                for datapoint in stats["Datapoints"]:
                    metric.unit = datapoint["Unit"]
                    break
            ret_val.append(metric)
        return ret_val

    def __get_send_items(self, stats, metric):
        send_items = []
        datapoints = stats["Datapoints"]
        datapoints = sorted(datapoints, key=lambda datapoints: datapoints["Timestamp"], reverse=True)
        for datapoint in datapoints:
            servicename = ''
            send_json_string = '{"host":"", "key":"", "value":"", "clock":""}'
            send_item = json.loads(send_json_string)

            if self.hostname == "undefined":
                send_item["host"] = self.identity
            else:
                send_item["host"] = self.hostname

            if self.service == "billing":
                for dimension in metric.dimensions:
                    if dimension["Name"] == "ServiceName":
                        servicename = dimension["Value"]
                send_item["key"] = 'cloudwatch.metric[%s.%s]' % (metric.name, servicename)
            else:
                send_item["key"] = 'cloudwatch.metric[%s]' % metric.name
            send_item["value"] = self.__get_datapoint_value_string(datapoint)
            send_item["clock"] = calendar.timegm(datapoint["Timestamp"].utctimetuple())
            send_items.append(send_item)
            break
        return send_items

    def __get_datapoint_value_string(self, datapoint):
        if "Average" in datapoint:
            return str(datapoint["Average"])
        elif "Sum" in datapoint:
            return str(datapoint["Sum"])
        elif "Maximum" in datapoint:
            return str(datapoint['Maximum'])
        else:
            return ""

    def __send_to_zabbix(self, send_data):
        send_data_string = json.dumps(send_data)
        zbx_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            zbx_client.connect((self.zabbix_host, self.zabbix_port))
        except Exception:
            print("Can't connect to zabbix server")
            quit()

        header = struct.pack('<4sBQ', 'ZBXD', 1, len(send_data_string))
        send_data_string = header + send_data_string
        try:
            zbx_client.sendall(send_data_string)
        except Exception:
            print('Data sending failure')
            quit()
        response = ''
        while True:
            data = zbx_client.recv(4096)
            if not data:
                break
            response += data

        print(response[13:])
        zbx_client.close()


    def send_metric_data_to_zabbix(self):
        now = "%.9f" % time.time()
        sec = now.split(".")[0]
        ns = now.split(".")[1]
        send_data = json.loads('{"request":"sender data","data":[],"clock":"%s","ns":"%s" }' % (sec, ns))
        metric_list = self.__get_metric_list()
        all_metric_stats = []
        servicename = self.service
        for metric in metric_list:
            if self.service == "billing":
                for dimension in metric.dimensions:
                    if dimension["Name"] == "ServiceName":
                        servicename = dimension["Value"]
            target_metric_info = {'namespace': metric.namespace, 'metricname': metric.name}
            for sum_stat_metric in self.sum_stat_metrics:  # for support each region metrics (RequestCount, RequestCount.ap-northeast-1 etc.)
                if metric.name.find(sum_stat_metric['metricname']) == 0:  # Only convert when finding the begging of string.
                    target_metric_info['metricname'] = sum_stat_metric['metricname']
            if target_metric_info in self.sum_stat_metrics:
                stats = self.__get_metric_stats(metric.name, metric.namespace, servicename, self.timerange_min, 'Sum', metric.dimensions)
            elif target_metric_info in self.max_stat_metrics:
                stats = self.__get_metric_stats(metric.name, metric.namespace, servicename, self.timerange_min, 'Maximum', metric.dimensions)
            else:
                stats = self.__get_metric_stats(metric.name, metric.namespace, servicename, self.timerange_min, metric.dimensions)
            send_data["data"].extend(self.__get_send_items(stats, metric))
        self.__send_to_zabbix(send_data)

    def show_metriclist_lld(self):
        lld_output_json = json.loads('{"data":[]}')
        metric_list = self.__get_metric_list()
        metric_list = self.__set_unit(metric_list)
        for metric in metric_list:
            lld_json_string = '{"{#METRIC.NAME}":"", "{#METRIC.UNIT}":"", "{#METRIC.NAMESPACE}":""}'
            lld_item = json.loads(lld_json_string)
            lld_item["{#METRIC.NAME}"] = metric.name
            lld_item["{#METRIC.NAMESPACE}"] = metric.namespace
            lld_item["{#METRIC.UNIT}"] = metric.unit
            lld_output_json["data"].append(lld_item)
            if self.service == "billing":
                lld_item["{#METRIC.SERVICENAME}"] = ""
                for dimension in metric.dimensions:
                    if dimension["Name"] == "ServiceName":
                        lld_item["{#METRIC.SERVICENAME}"] = dimension["Value"]
        print(json.dumps(lld_output_json))


FILE_NAME = 'date_time.txt'
LOG_DELAY_MS = 30000


class LogAwsZabbix(AwsZabbix):

    def __init__(self, region, access_key, secret, hostname, timerange_min,
                 zabbix_host, zabbix_port, log_group, log_stream):
        self.hostname = hostname
        self.log_group = log_group
        self.log_stream = log_stream
        self.zabbix_host = zabbix_host
        self.zabbix_port = zabbix_port
        self.timerange_min = timerange_min
        self.client_log = boto3.client(
            'logs',
            region_name=region,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret
        )

    def __get_logs(self):
        path = os.getcwd() + '/' + FILE_NAME
        timerange = self.timerange_min * 60000
        if os.path.exists(path):
            with open(FILE_NAME, 'r+') as file:
                start_time = int(file.read())
                end_time = start_time + timerange
                file.seek(0)
                file.write(str(end_time))
        else:
            with open(FILE_NAME, 'w') as file:
                end_time = int(time.time() * 1000 - LOG_DELAY_MS)
                start_time = end_time - timerange
                file.write(str(end_time))
        log_json_data = self.client_log.get_log_events(
            logGroupName=self.log_group,
            logStreamName=self.log_stream,
            startTime=start_time,
            endTime=end_time,
        )
        log_list = []
        for event in log_json_data['events']:
            log_list.append(event)
        return log_list

    def __get_send_items(self, log_list):
        send_items = []
        for log in log_list:
            send_json_string = '{"host":"", "key":"", "value":"", "clock":""}'
            send_item = json.loads(send_json_string)

            send_item["host"] = self.hostname
            send_item["key"] = 'cloudwatch.logs[{}:{}]'.format(self.log_group, self.log_stream)
            send_item["value"] = log['message']
            send_item["clock"] = int(log['timestamp'] / 1000)
            send_items.append(send_item)
        return send_items

    def show_log_data(self):
        log_list = self.__get_logs()
        print(log_list)

    def send_log_data_to_zabbix(self):
        now = "%.9f" % time.time()
        sec = now.split(".")[0]
        ns = now.split(".")[1]
        send_data = json.loads('{"request":"sender data","data":[],"clock":"%s","ns":"%s" }' % (sec, ns))
        log_list = self.__get_logs()
        send_data["data"].extend(self.__get_send_items(log_list))
        # self.__send_to_zabbix(send_data)
        print(send_data)

get_metrics.py

import argparse
from aws_zabbix.aws_zabbix import AwsZabbix
import os
import time

# identityはPodNameを入力
# serviceはeks
# containerの場合は別途考える必要あり

parser = argparse.ArgumentParser(description='Get AWS CloudWatch Metric list json format.')
parser.add_argument('-r', '--region', default=os.getenv("AWS_DEFAULT_REGION"), help='set AWS region name(e.g.: ap-northeast-1)')
parser.add_argument('-a', '--accesskey', default=os.getenv("AWS_ACCESS_KEY_ID"), help='set AWS Access Key ID')
parser.add_argument('-s', '--secret', default=os.getenv("AWS_SECRET_ACCESS_KEY"), help='set AWS Secret Access Key')
parser.add_argument('-i', '--identity', required=True, help='set Identity data (ec2: InstanceId, elb: LoadBalancerName, rds: DBInstanceIdentifier, ebs: VolumeId, eks: PodName)')
parser.add_argument('-H', '--hostname', default='undefined', help='set string that has to match HOST.HOST. defaults to identity)')
parser.add_argument('-m', '--send-mode', default='False', help='set True if you send statistic data (e.g.: True or False)')
parser.add_argument('-t', '--timerange', type=int, default=10, help='set Timerange min')
parser.add_argument('-p', '--zabbix-port', type=int, default=10051, help='set listening port number for Zabbix server')
parser.add_argument('-z', '--zabbix-host', default='localhost', help='set listening IP address for Zabbix server')
parser.add_argument('service', metavar='service_name', help='set Service name (e.g.: ec2 or elb or rds or eks')

args = parser.parse_args()
aws_zabbix = AwsZabbix(region=args.region, access_key=args.accesskey, secret=args.secret,
                       identity=args.identity, hostname=args.hostname, service=args.service,
                       timerange_min=args.timerange, zabbix_host=args.zabbix_host, zabbix_port=args.zabbix_port)

if args.send_mode.upper() == 'TRUE':
    aws_zabbix.send_metric_data_to_zabbix()
else:
    aws_zabbix.show_metriclist_lld()

get_log.py

import argparse
from aws_zabbix.aws_zabbix import LogAwsZabbix
import os
import time


parser = argparse.ArgumentParser(description='Get AWS CloudWatch Metric list json format.')
parser.add_argument('-r', '--region', default=os.getenv("AWS_DEFAULT_REGION"), help='set AWS region name(e.g.: ap-northeast-1)')
parser.add_argument('-a', '--accesskey', default=os.getenv("AWS_ACCESS_KEY_ID"), help='set AWS Access Key ID')
parser.add_argument('-s', '--secret', default=os.getenv("AWS_SECRET_ACCESS_KEY"), help='set AWS Secret Access Key')
parser.add_argument('-H', '--hostname', default='undefined', help='set string that has to match HOST.HOST. defaults to identity)')
parser.add_argument('-m', '--send-mode', default='False', help='set True if you send log-event data (e.g.: True or False)')
parser.add_argument('-t', '--timerange', type=int, default=10, help='set Timerange min')
parser.add_argument('-p', '--zabbix-port', type=int, default=10051, help='set listening port number for Zabbix server')
parser.add_argument('-z', '--zabbix-host', default='localhost', help='set listening IP address for Zabbix server')
parser.add_argument('-LG', '--log-group', required=True, help='set log group')
parser.add_argument('-LS', '--log-stream', required=True, help='set log stream')
args = parser.parse_args()
log_aws_zabbix = LogAwsZabbix(region=args.region, access_key=args.accesskey, secret=args.secret,
                       hostname=args.hostname, timerange_min=args.timerange,
                       zabbix_host=args.zabbix_host, zabbix_port=args.zabbix_port,
                       log_group=args.log_group, log_stream=args.log_stream)

if args.send_mode.upper() == 'TRUE':
    log_aws_zabbix.send_log_data_to_zabbix()
else:
    log_aws_zabbix.show_log_data()

内容

これまでの流れで、入力のテキストデータを分割したものをノードに、分割したもの同士の母音の一致をエッジの重みにしてネットワーク分析を行ってみる。グラフの描写、中心性を見るところまでが目標。

データを分割し、辞書を作る

from pykakasi import kakasi
import re
from collections import defaultdict
from janome.tokenizer import Tokenizer

with open("./gennama.txt","r") as f:
    data = f.read()

tokenizer = Tokenizer()
tokens = tokenizer.tokenize(data)

surface_list = []
part_of_speech_list = []
for token in tokens:
    surface_list.append(token.surface)
    part_of_speech_list.append(token.part_of_speech.split(",")[0])

segment_text = []
for i in range(len(surface_list)):
    if part_of_speech_list[i] == "記号":
        continue
    elif part_of_speech_list[i] == "助詞" or part_of_speech_list[i] == "助動詞":
        row = segment_text.pop(-1) + surface_list[i]
    else:
        row = surface_list[i]
    segment_text.append(row)

kakasi = kakasi()

kakasi.setMode('H', 'a')
kakasi.setMode('K', 'a')
kakasi.setMode('J', 'a')

conv = kakasi.getConverter()
text_data = [conv.do(text) for text in segment_text]
vowel_data = [re.sub(r"[^aeiou]+","",text) for text in text_data]
#{0:"oea"}
dic_vo = {k:v for k,v in enumerate(vowel_data)}
#voel_dataのインデックスで母音変換前のdataが分かるように辞書作成。{0:"俺は"}
dic = {k:v for k,v in enumerate(mecab_text)}

part3でやったことを活用。今回はN-gramは向いていないと判断。この、dic_voのキーの数だけノードがあり、ノード間は母音の一致があるかどうかで繋がりを見る。母音が一致する長さが長いほど重みを付けるようにしておく。part1で作っていたような方法を使用するが、自身への繋がりと２文字以上の一致からエッジが出来るようにする。

グラフを作る

#dic_voを渡し、インデックスがnode,値がedge,重みがscoreの(node,node,score)を作る。
def create_edge(dic_vo):
    node_len = len(dic_vo)

    edge_list = []
    for i in range(node_len):
        for j in range(node_len):
            score = create_weight(dic_vo[i],dic_vo[j])
            if score != 0:
                edge_list.append((i,j,score))
    return edge_list

def create_weight(word_a, word_b):
    weight = 0
    if len(word_a) > len(word_b):
        max_len = len(word_b)
        for i in range(max_len):
            for j in range(max_len + 1):
                if word_b[i:j] in word_a:
                    if word_b == word_a:
                        continue
                    elif len(word_b[i:j]) < 2:
                        continue
                    else:
                        weight += len(word_b[i:j])
    else:
        max_len = len(word_a)
        for i in range(max_len):
            for j in range(max_len + 1):
                if word_a[i:j] in word_b:
                    if word_a == word_b:
                        continue
                    elif len(word_b[i:j]) < 2:
                        continue
                    else:
                        weight += len(word_a[i:j])
    return weight

edge_list = create_edge(dic_vo)            

あとはこのedge_listをもとにグラフを描写する。ついでに、固有ベクトル中心性と媒介中心性が高いノードを取得し、元のデータを表示してみる。

import networkx as nx
import matplotlib.pyplot as plt

G = nx.Graph()
G.add_weighted_edges_from(edge_list)
pos = nx.spring_layout(G)
nx.draw_networkx_edges(G, pos)

plt.show()

#固有ベクトル中心性
cent = nx.eigenvector_centrality_numpy(G)
max_cent_node = max(list(cent.keys()), key=lambda val: cent[val])
#媒介中心性
between_cent = nx.communicability_betweenness_centrality(G)
max_betw_node = max(list(between_cent.keys()), key=lambda val: between_cent[val])

print("固有ベクトル中心性が高いのは:" + dic[max_cent_node])
print("媒介中心性が高いのは:" + dic[max_betw_node])

結果は予想通り、part2で行った「target_wordを絞れるのでは」と同じだ。まあ同じことをしているので当たり前だが、networkxではこのグラフを元にまだ出来ることがありそうなので追及してみる。

今後の方針

スコアの付け方で、「い」と「う」に注目し、前の音が「e」と「o」ならば、つまり「eい」「oう」ならば「ee」「oo」と変換して母音の一致を見るというものを考えている。（外来語の発音を参考にした）日本語でもそれは聞き分けにくく、響きは同じだと言えるはずだ。ラップ界での押韻の扱いではNGだと思われるがやってみる。ちなみに皆さんは本場の「ABCの歌」を聞いたことがあるだろうか？「LMNOP」を「エレネノピー」と一気にいくあれだ。子供の頃から「韻」が身近にあるようだ。日本語ラップへのリスペクトは忘れないが、韻を少し拡張して捉えることを試みる