240 lines
9.0 KiB
Python
240 lines
9.0 KiB
Python
import math
|
||
import numpy as np
|
||
from sklearn.svm import SVR
|
||
from sklearn.linear_model import Ridge
|
||
|
||
import requests
|
||
|
||
|
||
class Predictor1:
|
||
def __init__(self,
|
||
borderline=50, # 当考生分数与录取线相差borderline时，即录取率大于99%
|
||
sigma=0.0, # 当考生分数等于录取线时，概率为50%，增加simga则录取概率增加，反之减小
|
||
mode='svr', # 'diff':差分法， 'svr'：机器学习
|
||
remove_edges=False, # 是否除去最大最小值
|
||
sample_range=4, # 使用最近几年的数据
|
||
keep_latest_data=True, # 是否始终使用上一年的数据
|
||
# 以下是svr模型的参数
|
||
kernel='rbf',
|
||
degree=3,
|
||
gamma='scale',
|
||
coef0=0.0,
|
||
tol=1e-3,
|
||
C=10.0,
|
||
epsilon=0.1,
|
||
shrinking=True,
|
||
cache_size=100,
|
||
verbose=False,
|
||
max_iter=-1):
|
||
'''
|
||
:param borderline: the probability of a score which is lager than borderline will be lager than 99%
|
||
:param sigma: control the middle line of probability. when the score equals to the passing score, the
|
||
probability will be 50%, if increasing sigma, the probability will larger than 50% otherwise
|
||
smaller than 50%
|
||
:param mode: 'diff' or 'svr
|
||
:param remove_edges: whether remove the max value and min value
|
||
:param sample_range: the number of data which is used to train
|
||
:param keep_latest_data: whether keep the data of the latest year
|
||
:param C: Regularization parameter
|
||
:param cache_size: Specify the size of the kernel cache (in MB)
|
||
'''
|
||
|
||
assert sample_range > 0 and borderline > 0
|
||
|
||
self.scale = 5. / borderline
|
||
self.sigma = sigma
|
||
self.mode = mode
|
||
|
||
self.remove_edges = remove_edges
|
||
self.sample_range = sample_range
|
||
self.keep_latest_data = keep_latest_data
|
||
|
||
self.kernel = kernel
|
||
self.degree = degree
|
||
self.gamma = gamma
|
||
self.coef0 = coef0
|
||
self.tol = tol
|
||
self.C = C
|
||
self.epsilon = epsilon
|
||
self.shrinking = shrinking
|
||
self.cache_size = cache_size
|
||
self.verbose = verbose
|
||
self.max_iter = max_iter
|
||
|
||
self.model_dict = {}
|
||
self.diff_dict = {}
|
||
|
||
def create_model(self):
|
||
return SVR(kernel=self.kernel,
|
||
degree=self.degree,
|
||
gamma=self.gamma,
|
||
coef0=self.coef0,
|
||
tol=self.tol,
|
||
C=self.C,
|
||
epsilon=self.epsilon,
|
||
shrinking=self.shrinking,
|
||
cache_size=self.cache_size,
|
||
verbose=self.verbose,
|
||
max_iter=self.max_iter)
|
||
|
||
def reset_models(self):
|
||
self.model_dict = {}
|
||
|
||
def modified_sigmoid(self, x):
|
||
return 1. / (1. + math.exp(-((x + self.sigma) * self.scale)))
|
||
|
||
def train(self, infos, data):
|
||
'''
|
||
:param infos: tuple (school, province, wenli, batch)
|
||
:param data: list[list] [[year, pro_cro, min_score]]
|
||
'''
|
||
|
||
assert isinstance(infos, tuple) and isinstance(data, list) and len(data) > 0
|
||
|
||
data.sort(key=lambda x: x[0], reverse=True)
|
||
|
||
if self.mode == 'svr':
|
||
if infos not in self.model_dict:
|
||
model = self.create_model()
|
||
self.model_dict[infos] = model
|
||
else:
|
||
model = self.model_dict[infos]
|
||
|
||
data = data[:self.sample_range]
|
||
train_x, train_y = [[d[0], d[1]] for d in data], [d[2] for d in data]
|
||
model.fit(train_x, train_y)
|
||
|
||
elif self.mode == 'diff':
|
||
N = self.sample_range + 2 if self.remove_edges else self.sample_range
|
||
diffs = [data[i][2] - data[i][1] for i in range(min(N, len(data)))]
|
||
|
||
if len(data) < N or not self.remove_edges:
|
||
self.diff_dict[infos] = sum(diffs) / len(diffs)
|
||
else:
|
||
if self.keep_latest_data:
|
||
self.diff_dict[infos] = (sum(diffs) - max(diffs[1:]) - min(diffs[1:])) / (len(diffs) - 2)
|
||
else:
|
||
self.diff_dict[infos] = (sum(diffs) - max(diffs) - min(diffs)) / (len(diffs) - 2)
|
||
|
||
else:
|
||
raise ValueError('invalid mode')
|
||
|
||
def predict(self, infos, data):
|
||
'''
|
||
:param infos: tuple [school, province, wenli, batch]
|
||
:param data: list[list] [[year, pro_cro, score]]
|
||
'''
|
||
assert isinstance(infos, tuple) and isinstance(data, list)
|
||
|
||
if self.mode == 'svr':
|
||
if infos not in self.model_dict:
|
||
raise ValueError("the model hasn't been trained")
|
||
else:
|
||
model = self.model_dict[infos]
|
||
|
||
pred_x, scores = [[d[0], d[1]] for d in data], [d[2] for d in data]
|
||
|
||
lines = model.predict(pred_x)
|
||
probabilities = [self.modified_sigmoid(scores[i] - lines[i]) for i in range(len(lines))]
|
||
|
||
elif self.mode == 'diff':
|
||
if infos not in self.diff_dict:
|
||
raise ValueError("the difference hasn't been calculated")
|
||
else:
|
||
diff = self.diff_dict[infos]
|
||
|
||
probabilities = [self.modified_sigmoid(data[i][2] - (data[i][1] + diff)) for i in range(len(data))]
|
||
|
||
else:
|
||
raise ValueError('invalid mode')
|
||
|
||
return probabilities
|
||
|
||
|
||
class Predictor2:
|
||
def __init__(self, borderline=20, previour_years_num=3, alpha=100, sigma=3.1 + 0.09):
|
||
self.borderline = borderline # 当考生分数与录取线相差borderline时，即录取率大于99%
|
||
self.previour_years_num = previour_years_num # 使用过去几年的数据
|
||
self.alpha = alpha
|
||
self.scale = sigma / borderline
|
||
self.norm_arr = np.load('/Applications/DeskTop/公司项目/芳林公司项目/pytest/utils/probability/norm_arr.npy')
|
||
|
||
def __predict_lowest_rank(self, previous_data, year):
|
||
'''
|
||
:param previous_data: list[list] [[year, low_sort]]
|
||
:param year: int
|
||
:return:
|
||
'''
|
||
model = Ridge()
|
||
train_x, train_y = [[(x[0] % 100) * self.alpha] for x in previous_data], [x[1] for x in previous_data]
|
||
model.fit(train_x, train_y)
|
||
|
||
return model.predict([[(year % 100) * self.alpha]])[0]
|
||
|
||
def __query_score(self, province, wenli, weici, base_url='http://www.zhiyuanhelp.com/index.php/api/getweici?'):
|
||
url = base_url + 'province=%d&wenli=%d&weici=%d' % (province, wenli, weici)
|
||
try:
|
||
score = int(requests.get(url).text)
|
||
except:
|
||
raise ValueError('query_score: fail to query the score')
|
||
|
||
return score
|
||
|
||
def __calculate_probability(self, difference):
|
||
table = {-1: 37, 0: 45, 1: 55, 2: 63, 3: 73, 4: 80, 5: 96, 12: 96, 19: 99}
|
||
r=np.random.randn()
|
||
|
||
if difference < 0:
|
||
difference = -difference
|
||
positive = False
|
||
else:
|
||
positive = True
|
||
|
||
if difference > self.borderline:
|
||
return 1 if positive else 0
|
||
else:
|
||
difference *= self.scale
|
||
|
||
h, w = np.shape(self.norm_arr)
|
||
i, j = 1, 1
|
||
|
||
while i < h:
|
||
if (i == h - 1) or (self.norm_arr[i][0] <= difference and difference < self.norm_arr[i + 1][0]):
|
||
break
|
||
i += 1
|
||
difference -= self.norm_arr[i][0]
|
||
|
||
while j < w:
|
||
if (j == w - 1) or (self.norm_arr[0][j] <= difference and difference < self.norm_arr[0][j + 1]):
|
||
break
|
||
j += 1
|
||
|
||
return self.norm_arr[i][j] if positive else 1 - self.norm_arr[i][j]
|
||
|
||
def predict(self, province, wenli, previous_data, previous_lowest_score, current_year, current_ranks):
|
||
'''
|
||
:param province: int 省份代码
|
||
:param wenli: int 文理代码
|
||
:param previous_data: list[list] [[year, low_sort]] 要预测的学校前几年的年份和最低录取位次
|
||
:param previous_lowest_score: int 要预测的学校上一年的最低录取分数
|
||
:param current_year: int 要预测的年份（今年）
|
||
:param current_ranks: list[int] 要预测的学生位次（今年）,可输入多个
|
||
'''
|
||
previous_data = previous_data[:self.previour_years_num]
|
||
previous_data.sort(key=lambda x: x[0], reverse=True)
|
||
|
||
current_lowest_rank = self.__predict_lowest_rank(previous_data, current_year)
|
||
|
||
previous_lowest_rank = previous_data[0][1]
|
||
previous_ranks = [current_rank - current_lowest_rank + previous_lowest_rank for current_rank in current_ranks]
|
||
previous_scores = [self.__query_score(province, wenli, previous_rank) for previous_rank in previous_ranks]
|
||
|
||
probabilities = [self.__calculate_probability(previous_score - previous_lowest_score) for previous_score in
|
||
previous_scores]
|
||
|
||
return probabilities
|
||
|
||
|
||
# model = Predictor2()
|
||
# print(model.predict(844, 2, [[2016, 11231], [2017, 12310], [2018, 12111]], 561, 2019, [14056]))
|