Pythonを使って宝くじの予想スクリプトを作ってみました。｜うつ病のリハビリにも効果的？

はじめに

数字選択式宝くじ（ロト、ナンバーズ）についての話です。

うつ病で長い間休職しているのですが、リハビリを兼ねて、Pythonでロト６とナンバーズ４の予想スクリプトを作ってみました。

Pythonというのはプログラミング言語の一種で、簡単に色々なことができるのです。ネットで調べながら、宝くじの過去の当選番号を分析して、次回の予想番号を出すスクリプトを作りました。

試しにそれを実行して、出た番号を買ってみるということをしています。

ロト６予想スクリプト

# -*- coding: utf-8 -*-
"""loto6_predictions.ipynb

Automatically generated by Colaboratory.

ロト６の数字予測スクリプト
"""

import pandas as pd
import numpy as np
from mlxtend.frequent_patterns import apriori
from mlxtend.frequent_patterns import association_rules
from mlxtend.preprocessing import TransactionEncoder
from sklearn.preprocessing import MinMaxScaler
from sklearn.preprocessing import LabelEncoder
from keras.models import Sequential
from keras.layers import Dense, LSTM
from google.colab import drive
# import pymysql as pm
import datetime

# 余分なワーニングを非表示にする
import warnings
warnings.filterwarnings('ignore')

# データフレーム表示用関数
from IPython.display import display

# 表示オプション調整
# pandasでの浮動小数点の表示精度
pd.options.display.float_format = '{:.3f}'.format

# 時刻情報を取得
t_delta = datetime.timedelta(hours=9)
JST = datetime.timezone(t_delta, 'JST')

# Googleドライブをマウント
drive.mount('/content/drive')

# データの読み込み
data = pd.read_csv("/content/drive/MyDrive/Colab Notebooks/loto6_draw_result.csv", encoding="utf-8",index_col=0, parse_dates=[1])

# コネクションを張る。
# conn = pm.connect(host = '[db_host]', port = [db_port], user = '[db_user]', passwd = '[db_password]', db = '[db_name]')

# data = pd.read_sql_query("SELECT hold_times, draw_date, book_num1, book_num2, book_num3, book_num4, book_num5, book_num6, bonus_num1 FROM loto6_draw_result order by hold_times", conn,index_col=['hold_times'], parse_dates=['draw_date'])

# データベース項目名が指定されている場合、日本語名に項目名を置き換え
# data = data.rename(columns = {'hold_times':'開催回','draw_date':'抽選日','book_num1':'本数字１','book_num2':'本数字２','book_num3':'本数字３','book_num4':'本数字４','book_num5':'本数字５','book_num6':'本数字６','bonus_num1':'ボーナス数字１','set_sphere':'セット球'})

# 最新100回分を抽出
data = data.tail(100)

data.head()

data.tail()

# データ変換
# 本数字1から6までを取得
# data = data[["本数字１", "本数字２", "本数字３", "本数字４", "本数字５", "本数字６"]]
data = data.iloc[:, 1:7]

# data.head()

# 過去抽選番号に乱数を加減する。
# data_bk = data.copy()
# data = data.add(np.random.randint(-1,2,6))

# 数字の範囲外となった場合の調整
# data = data.mask(data > 43, data - 43)
# data = data.mask(data < 1, data + 43)

# ベクトル分布に変換
te = TransactionEncoder()
te_ary = te.fit(data.values).transform(data.values)
data_dataset = pd.DataFrame(te_ary, columns=te.columns_)

# baloon型 (True/False)→float型(1/0)に変換
data_dataset_1 = data_dataset * 1

lot_max_number = len(data_dataset.columns) # 抽選された最大の数字
lot_size = len(data.columns)  # 数字の個数
prev_numbers = list(data.iloc[-1,0:lot_size].values) # 直近の数字

# Aprioriアルゴリズムによるアソシエーション分析
frequent_itemsets = apriori(data_dataset, min_support=0.02, use_colnames=True)
rules = association_rules(frequent_itemsets, metric="confidence", min_threshold=1.0)

# 結果の出力(lift値で降順ソート)
rules.sort_values(by=['lift'],ascending=False).head(10)

# LSTMによる予測
# 入力データの作成
numbers = data_dataset_1.values
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_numbers = scaler.fit_transform(numbers)

# Split the data into training and testing sets
train_size = int(len(scaled_numbers) * 0.8)
train_data = scaled_numbers[0:train_size, :]
test_data = scaled_numbers[train_size:len(scaled_numbers), :]

# Prepare the training and testing data
def create_dataset(dataset, time_steps=1):
    X, Y = [], []
    for i in range(len(dataset) - time_steps - 1):
        a = dataset[i:(i + time_steps), :]
        X.append(a)
        Y.append(dataset[i + time_steps, :])
    return np.array(X), np.array(Y)

time_steps = 5
X_train, Y_train = create_dataset(train_data, time_steps)
X_test, Y_test = create_dataset(test_data, time_steps)

# モデルの構築
model = Sequential()
model.add(LSTM(128, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2])))
model.add(LSTM(128, return_sequences=True))
model.add(LSTM(128))
model.add(Dense(lot_max_number))
model.compile(loss='mean_squared_error', optimizer='adam')

# モデルのトレーニング
model.fit(X_train, Y_train, epochs=100, batch_size=32, validation_data=(X_test, Y_test))

# 予測の実行
last_sequence = scaled_numbers[-time_steps:, :]
last_sequence = last_sequence.reshape(1, time_steps, lot_max_number)
prediction = model.predict(last_sequence)
prediction_s = scaler.inverse_transform(prediction)

# 予測結果の行列変換
prediction_t = prediction_s.T

# 予測結果の表示
# print(next_nums)
for i, prediction_t  in enumerate(prediction_t):
    print(f"{i+1:02}：{prediction_t}")

# 結果の表示(期待値：降順)
prediction_d = pd.DataFrame(prediction_s)
prediction_d.columns = prediction_d.columns  + 1
prediction_d.sort_values(by=0,ascending=False, inplace=True, axis=1)
prediction_d.T.head(10)

# 予測結果を表示する
predicted_numbers  = list(prediction_d.iloc[0, 0:lot_size].index)

now = datetime.datetime.now(JST)
print('実行日時：',now.strftime('%Y/%m/%d %H:%M:%S'))
print('*** 前回に出現した数字 ***')
print(prev_numbers)
print('*** 次回に出現する数字 ***')
print(predicted_numbers)

# モンテカルロ法による予測
# 1から43までの数字の出現回数を計算する
freq = np.zeros(lot_max_number+1)
for i in range(lot_size):
    freq += np.bincount(data.iloc[:, i], minlength=lot_max_number+1)
freq = np.delete(freq, 0, 0)

# 出現回数に応じた確率分布を作成
prob = freq / freq.sum()

# 予測を行う
num_samples = 10000  # 生成するサンプル数
predictions = []
for i in range(num_samples):
    # 1から43までの数字を重みとして、ランダムに数字を選ぶ
    nums = np.random.choice(np.arange(1, lot_max_number + 1), size=lot_size, replace=False, p=prob)
    predictions.append(sorted(nums))
    # predictions.append(list(nums))

# 予測結果を表示する
now = datetime.datetime.now(JST)
print('実行日時：',now.strftime('%Y/%m/%d %H:%M:%S'))
print('*** 前回に出現した数字 ***')
print(prev_numbers)
print('*** 次回に出現する数字 ***')
print(predictions[num_samples - 5:])

# 予測シュミレーション（10000件）をDataframeに変換
columns = [f'num{i}' for i in range(1, lot_size+1)]
df_predictions = pd.DataFrame(predictions,columns=columns)
df_predictions.index.name="pred_times"
df_predictions.head()

df_predictions.tail()

# 予測シュミレーション（10000件）から条件を指定して抽出
any_col_is_1 = df_predictions[columns].eq(25).any(axis=1)
any_col_is_2 = df_predictions[columns].eq(1).any(axis=1)
any_col_is_3 = df_predictions[columns].eq(35).any(axis=1)
df_predictions[any_col_is_1 & any_col_is_2 & any_col_is_3]

# [EOF]

ナンバーズ４予想スクリプト

<!-- wp:paragraph -->
<p><span class="red">2023/4/26 出力部の表示、シュミレーション結果の検索を追加</span></p>
<!-- /wp:paragraph --># -*- coding: utf-8 -*-
"""numbers4_predictions.ipynb

Automatically generated by Colaboratory.

ナンバーズ４の数字予測スクリプト
"""

import pandas as pd
import numpy as np
from mlxtend.frequent_patterns import apriori
from mlxtend.frequent_patterns import association_rules
from mlxtend.preprocessing import TransactionEncoder
from sklearn.preprocessing import MinMaxScaler
from sklearn.preprocessing import LabelEncoder
from keras.models import Sequential
from keras.layers import Dense, LSTM
from google.colab import drive
# import pymysql as pm
import datetime

# 余分なワーニングを非表示にする
import warnings
warnings.filterwarnings('ignore')

# データフレーム表示用関数
from IPython.display import display

# 表示オプション調整
# pandasでの浮動小数点の表示精度
pd.options.display.float_format = '{:.3f}'.format

# 時刻情報を取得
t_delta = datetime.timedelta(hours=9)
JST = datetime.timezone(t_delta, 'JST')

# Googleドライブをマウント
drive.mount('/content/drive')

# データの読み込み
data = pd.read_csv("/content/drive/MyDrive/Colab Notebooks/numbers4_draw_result.csv", encoding="utf-8",index_col=0, parse_dates=[1],dtype={'当選数字':'str'})

# コネクションを張る。
# conn = pm.connect(host = '[db_host]', port = [db_port], user = '[db_user]', passwd = '[db_password]', db = '[db_name]')

# data = pd.read_sql_query("SELECT hold_times, draw_date, win_num, win_num_thou_vr, win_num_hund_vr, win_num_ten_vr, win_num_one_vr FROM numbers4_draw_result ORDER BY hold_times", conn,index_col=['hold_times'], parse_dates=['draw_date'],dtype={'win_num':str,'win_num_thou_vr':'int64','win_num_hund_vr':'int64','win_num_ten_vr':'int64','win_num_one_vr':'int64'})

# データベース項目名が指定されている場合、日本語名に項目名を置き換え
# data = data.rename(columns = {'hold_times':'開催回','draw_date':'抽選日','win_num':'当選数字','win_num_thou_vr':'当選数字（千の位）','win_num_hund_vr':'当選数字（百の位）','win_num_ten_vr':'当選数字（十の位）','win_num_one_vr':'当選数字（一の位）'})

# 最新100回分を抽出
data = data.tail(100)

data.head()

data.tail()

# データ変換
# 本数字1から8までを取得
# data = data[['当選数字（千の位）','当選数字（百の位）', '当選数字（十の位）', '当選数字（一の位）'] )]
data = data.iloc[:, 2:6]

# data.head()

# 過去抽選番号に乱数を加減する。
# data_bk = data.copy()
# data = data.add(np.random.randint(-1,2,4))
# data = data.mask(data > 9, data - 10)
# data = data.mask(data < 0, data + 10)

prev_numbers = list(data.iloc[-5:,0:4].values) # 直近5回の数字

# ベクトル表現版 前準備
# data_ts = pd.DataFrame(index=data.index, columns=['当選数字（千の位）','当選数字（百の位）', '当選数字（十の位）', '当選数字（一の位）'] )
data_ts = pd.DataFrame(index=data.index, columns=['win_num_thou_vr','win_num_hund_vr','win_num_ten_vr','win_num_one_vr']
data_ts.iloc[:, 0] = data.iloc[:, 0] # 当選数字（千の位） / 数字に１０を加算
data_ts.iloc[:, 1] = data.iloc[:, 1] + 10 # 当選数字（百の位）
data_ts.iloc[:, 2] = data.iloc[:, 2] + 20 # 当選数字（十の位） / 数字に2０を加算
data_ts.iloc[:, 3] = data.iloc[:, 3] + 30 # 当選数字（一の位） / 数字に3０を加算

# ベクトル分布に変換
te = TransactionEncoder()
te_ary = te.fit(data.values).transform(data.values)
data_dataset = pd.DataFrame(te_ary, columns=te.columns_)

te_ts = TransactionEncoder()
te_ts_ary = te_ts.fit(data_ts.values).transform(data_ts.values)
data_dataset_ts = pd.DataFrame(te_ts_ary, columns=te_ts.columns_)

# baloon型 (True/False)→float型(1/0)に変換
data_dataset_1 = data_dataset * 1
data_dataset_ts_1 = data_dataset_ts * 1

# Aprioriアルゴリズムによるアソシエーション分析
frequent_itemsets = apriori(data_dataset, min_support=0.02, use_colnames=True)
rules = association_rules(frequent_itemsets, metric="confidence", min_threshold=0.6)

# 結果の出力(lift値で降順ソート)
rules.sort_values(by=['lift'],ascending=False).head(10)

# LSTMによる予測
# 入力データの作成
numbers = data_dataset_ts_1.values
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_numbers = scaler.fit_transform(numbers)

# Split the data into training and testing sets
train_size = int(len(scaled_numbers) * 0.8)
train_data = scaled_numbers[0:train_size, :]
test_data = scaled_numbers[train_size:len(scaled_numbers), :]

# Prepare the training and testing data
def create_dataset(dataset, time_steps=1):
    X, Y = [], []
    for i in range(len(dataset) - time_steps - 1):
        a = dataset[i:(i + time_steps), :]
        X.append(a)
        Y.append(dataset[i + time_steps, :])
    return np.array(X), np.array(Y)

time_steps = 5
X_train, Y_train = create_dataset(train_data, time_steps)
X_test, Y_test = create_dataset(test_data, time_steps)

# モデルの構築
model = Sequential()
model.add(LSTM(128, return_sequences=True, input_shape=(X_train.shape[1], X_train.shape[2])))
model.add(LSTM(128, return_sequences=True))
model.add(LSTM(128))
model.add(Dense(40))
model.compile(loss='mean_squared_error', optimizer='adam')

# モデルのトレーニング
model.fit(X_train, Y_train, epochs=100, batch_size=32, validation_data=(X_test, Y_test))

# 予測の実行
last_sequence = scaled_numbers[-time_steps:, :]
last_sequence = last_sequence.reshape(1, time_steps, 40)
prediction = model.predict(last_sequence)
prediction_s = scaler.inverse_transform(prediction)

# 予測結果の行列変換
prediction_t = prediction_s.T

# 予測結果の表示
for i, prediction_t  in enumerate(prediction_t):
    print(f"{i:02}：{prediction_t}")

# 結果の表示(期待値：降順)
prediction_d = pd.DataFrame(prediction_s)
prediction_d.columns = prediction_d.columns
prediction_d.head()

# 各パート毎（0～9,･･･ , 30～39）に区分け
prediction_dp = []
for i in range(0, 40, 10):
    prediction_di = prediction_d.iloc[:, i:i+10]
    prediction_di.sort_values(by=0, ascending=False, inplace=True, axis=1)
    prediction_dp.append(prediction_di.columns)

# 数字全体について結果の表示(期待値：降順)
prediction_d.sort_values(by=0, ascending=False, inplace=True, axis=1)
prediction_d.T.head(15)

# 予測結果を表示する
predicted_num = []
for i in range(0, 4):
  predicted_num.append(prediction_dp[i][0])

# 計算のために付けたインデックスの十の位を削除（10で割った余りを設定）
calc_mod = lambda x : x % 10
predicted_num_list = list(map(calc_mod,predicted_num))

# 予測結果を表示する
now = datetime.datetime.now(JST)
print('実行日時：',now.strftime('%Y/%m/%d %H:%M:%S'))
print('*** 前回に出現した数字 ***')
print(np.array(prev_numbers))
print('*** 次回に出現する数字 ***')
print(predicted_num_list)

# モンテカルロ法による予測
# 0から39までの数字の出現回数を計算する
freq = np.zeros(40)
for i in range(4):
    freq += np.bincount(data_ts.iloc[:, i], minlength=40)

freq_list = np.array_split(freq, 4)

# 出現回数に応じた確率分布を作成
prob_list = []
for i in range(0, 4):
  prob_list.append(freq_list[i] / freq_list[i].sum())

# 予測を行う
num_samples = 10000  # 生成するサンプル数
predicted_num, predicted_num_tmp  = [],[]

for i in range(num_samples):
  for j in range(3,-1,-1):
    # 各桁毎に0から9までの数字を重みとして、ランダムに数字を選ぶ
    nums_start = (j  * 10)
    nums_end = ((j + 1)  * 10)
    nums = np.random.choice(np.arange(nums_start, nums_end), size=5, replace=True, p=prob_list[j])
    predicted_num_tmp.insert(0, nums)
    if j == 0:
      num_list = np.array(predicted_num_tmp).T[0]
      predicted_num.insert(0, num_list)
      predicted_num_tmp = []

predicted_num_list = list(map(calc_mod,predicted_num))

# 予測結果を表示する
now = datetime.datetime.now(JST)
print('実行日時：',now.strftime('%Y/%m/%d %H:%M:%S'))
print('*** 前回に出現した数字 ***')
print(np.array(prev_numbers))
print('*** 次回に出現する数字 ***')
print(np.array(predicted_num_list[0:5]))

# 予測シュミレーション（10000件）をDataframeに変換
columns = [f'num{i}' for i in range(1, 5)]
df_predictions = pd.DataFrame(predicted_num_list,columns=columns)
df_predictions.index.name="pred_times"
df_predictions.head()

df_predictions.tail()

# 予測シュミレーション（10000件）から条件を指定して抽出
any_col_is_1 = df_predictions[columns].eq(3).any(axis=1)
any_col_is_2 = df_predictions[columns].eq(6).any(axis=1)
any_col_is_3 = df_predictions['num1'].eq(5)
df_predictions[any_col_is_1 & any_col_is_2 & any_col_is_3]

# [EOF]

おわりに

もちろん、宝くじは確率的に当たる可能性は非常に低いので、なかなか当たりません。でも、プログラミングをすることで頭を使ったり、宝くじを買うことでワクワクしたりするので、うつ病のリハビリになっています。もし当たったら、それはさらに嬉しいことですね。当たるといいなぁ。