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]))