之前的帖子,是基于PRD算法去模拟。
现状是全新的算法使用的是机器学习中的逻辑回归算法,由于历史样本数量有限,才1250多次强化历史记录。所以现状算法效果能保证78% 准确率。
算法类主要逻辑是:1、 数据加载与清理 2、特征工程处理 3、模型训练、搜参、保存 4、模型预测
GUI 类的代码就不贴了,直接放百度网盘了。 exe程序需要和保存的模型在一个目录。
链接: https://pan.baidu.com/s/1pDblQGjIFxAYqAPdjC1yNQ?pwd=1314 提取码: 1314--来自百度网盘超级会员v2的分享
使用方法分析按钮会打出日志,概率不是真实的,是程序输出的,结果0表示失败1表示成功。可以切换强化等级或使用道具,多次点击分析按钮。 成功和失败按钮是表示已经执行了强化,让模拟器记录你过去的强化历史,为下一次分析预测更加准确。 另外可以将真实的强化历史 导出excel后,发给我,我积累数据后 重新训练模型,让算法模型更加准确。
import joblib
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import GridSearchCV, train_test_split
from sklearn.preprocessing import StandardScaler
# 装备强化算法
class EquipmentEnhancement:
def __init__(self):
# 不同等级的显示概率
self.T = {0: 1, 1: 1, 2: 0.9, 3: 0.8, 4: 0.7, 5: 0.6, 6: 0.5, 7: 0.4, 8: 0.3, 9: 0.2, 10: 0.2, 11: 0.2, 12: 0.2,
13: 0.2, 14: 0.2, 15: 0.15, 16: 0.14, 17: 0.13, 18: 0.12, 19: 0.11, 20: 0.10}
self.scaler = StandardScaler()
self.model = None
def data_load(self):
# 数据加载
samples = pd.read_excel('./samples.xlsx').query('动作=="强化"')
print(samples)
print(len(samples))
samples['历史'] = 1
return samples
def fe_engineer(self, df, train=True):
# 特征工程 ,其中 等级,强化等级,道具,批次号是固定输入的特征数据
samples_columns = [ '强化等级', '道具', '批次号', '结果', '历史']
X_columns = ['强化等级', '道具', '面板显示概率', '强化等级的概率', '强化等级的连续失败次数',
'全局概率',
'全局连续失败次数', '全局失败次数',
]
y_column = 'y'
df['强化等级'] = df['强化等级'].astype('int')
df['道具'] = df['道具'].astype('int')
df['批次号'] = df['批次号'].astype('int')
df['历史'] = df['历史'].astype('int')
# , '全局强化次数'
# , '强化等级的次数'
df = df[samples_columns].copy()
df[y_column] = 0
df.loc[df['结果'] == '成功', y_column] = 1
df['面板显示概率'] = df['强化等级'].map(self.T)
def consecutive_fails_before(s):
s = (s == 0).astype(int)
fail_streaks = s * (s.groupby((s != s.shift()).cumsum()).cumcount() + 1)
return fail_streaks.shift(fill_value=0)
# 初始化新字段
# 相同强化等级的成功概率
df['全局概率'] = df.groupby('批次号')[y_column].transform(
lambda x: x.expanding().mean().shift(fill_value=0))
# 相同强化等级的强化次数
df['全局强化次数'] = df.groupby('批次号').cumcount()
# 相同强化等级的连续失败次数
df['全局连续失败次数'] = df.groupby('批次号')[y_column].transform(consecutive_fails_before)
df['全局失败次数'] = df.groupby('批次号')[y_column].transform(
lambda x: (x == 0).expanding().sum().shift(fill_value=0))
df['全局成功次数'] = df.groupby('批次号')[y_column].transform(
lambda x: (x == 1).expanding().sum().shift(fill_value=0))
# 相同强化等级的成功概率
df['强化等级的概率'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: x.expanding().mean().shift(fill_value=0))
# 相同强化等级的强化次数
df['强化等级的次数'] = df.groupby(['批次号', '强化等级']).cumcount()
# 相同强化等级的连续失败次数
df['强化等级的连续失败次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
consecutive_fails_before)
# 相同强化等级的失败次数
df['强化等级的失败次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: (x == 0).expanding().sum().shift(fill_value=0))
# 相同强化等级的失败次数
df['强化等级的成功次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: (x == 1).expanding().sum().shift(fill_value=0))
if train:
self.scaler.fit(df[X_columns], df[y_column])
df[X_columns] = self.scaler.transform(df[X_columns])
train_X = df.query('历史==1')[X_columns]
# X_train_scaled = scaler.fit_transform(train_X)
train_y = df.query('历史==1')[y_column]
pred_X = df.query('历史==0')[X_columns]
return train_X, train_y, pred_X
def train(self, X, y):
# 训练和保存模型
# 分割数据为训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 定义参数网格
param_grid = {
'C': np.logspace(-4, 4, 10), # 对于C,我们尝试一个从10^-4到10^4的对数空间
'penalty': ['l1', 'l2'], # 尝试L1和L2正则化
'solver': ['liblinear', 'saga'] # 由于L1正则化与所有求解器都不兼容,我们选择这两个
}
# 使用逻辑回归模型进行训练
clf = LogisticRegression(max_iter=10000)
# 初始化GridSearchCV
grid_search = GridSearchCV(clf, param_grid, cv=5, verbose=1)
# 进行搜索
grid_search.fit(X_train, y_train)
# grid_search.fit(X, y)
# 打印最佳参数
print("Best parameters found: ", grid_search.best_params_)
# 使用最佳参数在测试集上进行评估
self.model = grid_search.best_estimator_
# 在测试集上进行预测
y_pred = self.model.predict(X_test)
# 预测成功的概率
# y_prob = best_clf.predict_proba(X_test)[:, 1] # [:, 1] 代表的是成功的概率
print("Accuracy:", accuracy_score(y_test, y_pred))
joblib.dump(self, 'best_model.joblib')
def pred(self, pred_X):
y_pred = self.model.predict(pred_X)
y_prob = self.model.predict_proba(pred_X)[:, 1] # [:, 1] 代表的是成功的概率
pred_X['预测结果'] = y_pred
pred_X['预测概率'] = y_prob
return pred_X
if __name__ == "__main__":
enhancement = EquipmentEnhancement()
df = enhancement.data_load()
X, y, pred_X = enhancement.fe_engineer(df)
enhancement.train(X, y)
现状是全新的算法使用的是机器学习中的逻辑回归算法,由于历史样本数量有限,才1250多次强化历史记录。所以现状算法效果能保证78% 准确率。
算法类主要逻辑是:1、 数据加载与清理 2、特征工程处理 3、模型训练、搜参、保存 4、模型预测
GUI 类的代码就不贴了,直接放百度网盘了。 exe程序需要和保存的模型在一个目录。
链接: https://pan.baidu.com/s/1pDblQGjIFxAYqAPdjC1yNQ?pwd=1314 提取码: 1314--来自百度网盘超级会员v2的分享
使用方法分析按钮会打出日志,概率不是真实的,是程序输出的,结果0表示失败1表示成功。可以切换强化等级或使用道具,多次点击分析按钮。 成功和失败按钮是表示已经执行了强化,让模拟器记录你过去的强化历史,为下一次分析预测更加准确。 另外可以将真实的强化历史 导出excel后,发给我,我积累数据后 重新训练模型,让算法模型更加准确。
import joblib
import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import GridSearchCV, train_test_split
from sklearn.preprocessing import StandardScaler
# 装备强化算法
class EquipmentEnhancement:
def __init__(self):
# 不同等级的显示概率
self.T = {0: 1, 1: 1, 2: 0.9, 3: 0.8, 4: 0.7, 5: 0.6, 6: 0.5, 7: 0.4, 8: 0.3, 9: 0.2, 10: 0.2, 11: 0.2, 12: 0.2,
13: 0.2, 14: 0.2, 15: 0.15, 16: 0.14, 17: 0.13, 18: 0.12, 19: 0.11, 20: 0.10}
self.scaler = StandardScaler()
self.model = None
def data_load(self):
# 数据加载
samples = pd.read_excel('./samples.xlsx').query('动作=="强化"')
print(samples)
print(len(samples))
samples['历史'] = 1
return samples
def fe_engineer(self, df, train=True):
# 特征工程 ,其中 等级,强化等级,道具,批次号是固定输入的特征数据
samples_columns = [ '强化等级', '道具', '批次号', '结果', '历史']
X_columns = ['强化等级', '道具', '面板显示概率', '强化等级的概率', '强化等级的连续失败次数',
'全局概率',
'全局连续失败次数', '全局失败次数',
]
y_column = 'y'
df['强化等级'] = df['强化等级'].astype('int')
df['道具'] = df['道具'].astype('int')
df['批次号'] = df['批次号'].astype('int')
df['历史'] = df['历史'].astype('int')
# , '全局强化次数'
# , '强化等级的次数'
df = df[samples_columns].copy()
df[y_column] = 0
df.loc[df['结果'] == '成功', y_column] = 1
df['面板显示概率'] = df['强化等级'].map(self.T)
def consecutive_fails_before(s):
s = (s == 0).astype(int)
fail_streaks = s * (s.groupby((s != s.shift()).cumsum()).cumcount() + 1)
return fail_streaks.shift(fill_value=0)
# 初始化新字段
# 相同强化等级的成功概率
df['全局概率'] = df.groupby('批次号')[y_column].transform(
lambda x: x.expanding().mean().shift(fill_value=0))
# 相同强化等级的强化次数
df['全局强化次数'] = df.groupby('批次号').cumcount()
# 相同强化等级的连续失败次数
df['全局连续失败次数'] = df.groupby('批次号')[y_column].transform(consecutive_fails_before)
df['全局失败次数'] = df.groupby('批次号')[y_column].transform(
lambda x: (x == 0).expanding().sum().shift(fill_value=0))
df['全局成功次数'] = df.groupby('批次号')[y_column].transform(
lambda x: (x == 1).expanding().sum().shift(fill_value=0))
# 相同强化等级的成功概率
df['强化等级的概率'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: x.expanding().mean().shift(fill_value=0))
# 相同强化等级的强化次数
df['强化等级的次数'] = df.groupby(['批次号', '强化等级']).cumcount()
# 相同强化等级的连续失败次数
df['强化等级的连续失败次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
consecutive_fails_before)
# 相同强化等级的失败次数
df['强化等级的失败次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: (x == 0).expanding().sum().shift(fill_value=0))
# 相同强化等级的失败次数
df['强化等级的成功次数'] = df.groupby(['批次号', '强化等级'])[y_column].transform(
lambda x: (x == 1).expanding().sum().shift(fill_value=0))
if train:
self.scaler.fit(df[X_columns], df[y_column])
df[X_columns] = self.scaler.transform(df[X_columns])
train_X = df.query('历史==1')[X_columns]
# X_train_scaled = scaler.fit_transform(train_X)
train_y = df.query('历史==1')[y_column]
pred_X = df.query('历史==0')[X_columns]
return train_X, train_y, pred_X
def train(self, X, y):
# 训练和保存模型
# 分割数据为训练集和测试集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
# 定义参数网格
param_grid = {
'C': np.logspace(-4, 4, 10), # 对于C,我们尝试一个从10^-4到10^4的对数空间
'penalty': ['l1', 'l2'], # 尝试L1和L2正则化
'solver': ['liblinear', 'saga'] # 由于L1正则化与所有求解器都不兼容,我们选择这两个
}
# 使用逻辑回归模型进行训练
clf = LogisticRegression(max_iter=10000)
# 初始化GridSearchCV
grid_search = GridSearchCV(clf, param_grid, cv=5, verbose=1)
# 进行搜索
grid_search.fit(X_train, y_train)
# grid_search.fit(X, y)
# 打印最佳参数
print("Best parameters found: ", grid_search.best_params_)
# 使用最佳参数在测试集上进行评估
self.model = grid_search.best_estimator_
# 在测试集上进行预测
y_pred = self.model.predict(X_test)
# 预测成功的概率
# y_prob = best_clf.predict_proba(X_test)[:, 1] # [:, 1] 代表的是成功的概率
print("Accuracy:", accuracy_score(y_test, y_pred))
joblib.dump(self, 'best_model.joblib')
def pred(self, pred_X):
y_pred = self.model.predict(pred_X)
y_prob = self.model.predict_proba(pred_X)[:, 1] # [:, 1] 代表的是成功的概率
pred_X['预测结果'] = y_pred
pred_X['预测概率'] = y_prob
return pred_X
if __name__ == "__main__":
enhancement = EquipmentEnhancement()
df = enhancement.data_load()
X, y, pred_X = enhancement.fe_engineer(df)
enhancement.train(X, y)