蓝球的代码基本完成

orm/blue_forecast.py

@@ -2,23 +2,16 @@ from db import *
 from util import *
 
 
-# last_data = get_last_data()
-# print(last_data)
-#
-# date_id_data = get_data_by_date_id('2023068')
-# print(date_id_data)
-
 # 获取全部数据
 def get_all_data():
     return db_get_all_data()
 
 
 # 获取所有篮球的统计数据
-def get_all_blue():
+def get_all_blue(_all_data):
     all_blue = dict()
-    all_data = db_get_all_data()
-    if all_data is not False:
-        for data in all_data:
+    if _all_data is not False:
+        for data in _all_data:
             if data['blue'] in all_blue.keys():
                 all_blue[data['blue']] += 1
             else:
@@ -30,21 +23,10 @@ def get_all_blue():
 
 # 获取蓝球统计次数与平均值的额偏差
 def get_delta_blue(_blue_data):
-    total = 0
-    if _blue_data is not False:
-        for _num in _blue_data.values():
-            total += _num
-        average = total / len(_blue_data.keys())
-        print(average)
-        _delta_blue_dict = dict()
-        for _key, _val in _blue_data.items():
-            _delta_blue_dict[_key] = _blue_data[_key] - average
-        return _delta_blue_dict
-
-
-print(get_delta_blue(get_all_blue()))
+    return dict_sort(get_analysis_data(_blue_data), 'val', True)
 
 
+# 获取全部向前追溯的蓝球数据列表
 def get_all_prv_blue_list(_all_data, _last_blue_list, _num):
     all_prv_blue_list = []
     min_id, max_id = 1, _all_data[-1]['id']  # 获取最大和最小id
@@ -59,17 +41,62 @@ def get_all_prv_blue_list(_all_data, _last_blue_list, _num):
     return sorted(set(all_prv_blue_list))
 
 
-# 获取预测的数据
-def get_blue_forecast(_all_data, _method, _num):
-    last_blue = db_get_last_data()['blue']
-    if last_blue is False:
-        return False
+# 获取全部向后预测的历史id列表
+def get_all_post_id_list(_all_data, _blue_id_list, _num):
+    post_id_list = []
+    min_id, max_id = 1, _all_data[-1]['id']  # 获取最大和最小id
+    for _id in _blue_id_list:
+        for n in range(1, _num + 1):
+            _post_id = _id + n
+            if min_id <= _post_id <= max_id:
+                post_id_list.append(_post_id)
+    return post_id_list
+
+
+# 获取蓝球统计数据
+def get_blue_statistics(_prv_blue_id_list):
+    _blue_data = db_get_data_in_ids(_prv_blue_id_list)
+    _blue_dict = dict()
+    for _data in _blue_data:
+        if _data['blue'] in _blue_dict.keys():
+            _blue_dict[_data['blue']] += 1
+        else:
+            _blue_dict[_data['blue']] = 1
+    return _blue_dict
+
+
+# 获取分析后的差异数据
+def get_analysis_data(_raw_blue_data):
+    total = 0
+    for _num in _raw_blue_data.values():
+        total += _num
+    average = total / len(_raw_blue_data.keys())
+    _delta_blue_dict = dict()
+    for _key, _val in _raw_blue_data.items():
+        _delta_blue_dict[_key] = round(_raw_blue_data[_key] - average, 2)
+    return _delta_blue_dict
+
+
+# 根据蓝球获取全部历史中出现过此蓝球的id列表
+def get_blue_id_list_by_blue(_all_data, _last_blue):
     blue_id_list = []
     for _data in _all_data:
-        if _data['blue'] == last_blue:
+        if _data['blue'] == _last_blue:
             blue_id_list.append(_data['id'])
-    print(blue_id_list)
-    min_id, max_id = 1, _all_data[-1]['id']  # 获取最大和最小id
+    return blue_id_list
+
+
+# 获取预测的数据
+def get_blue_forecast(_all_data, _method, _num):
+    # last_blue = db_get_last_data()['blue']
+    # if last_blue is False:
+    #     return False
+    # blue_id_list = []
+    # for _data in _all_data:
+    #     if _data['blue'] == last_blue:
+    #         blue_id_list.append(_data['id'])
+    # # print(blue_id_list)
+    # min_id, max_id = 1, _all_data[-1]['id']  # 获取最大和最小id
     """
     当_num=3时
     先取当前一期的蓝球号码，然后计算历史上相同蓝球出现的后一期蓝球
@@ -79,13 +106,32 @@ def get_blue_forecast(_all_data, _method, _num):
 
     if _method == 'prv':  # 当前一期往前推num期，包含此期数据
         recent_blue_id_list = [d['blue'] for d in _all_data[-_num:]]
-        prv_blue_id_list = get_all_prv_blue_list(_all_data, recent_blue_id_list, _num)
-        print(prv_blue_id_list)
+        prv_blue_id_list = get_all_prv_blue_list(_all_data, recent_blue_id_list, _num)  # 根据蓝球出现的历史记录查询出向前推算的蓝球对应数据id号
+        _raw_blue_data = get_blue_statistics(prv_blue_id_list)  # 根据上步查出的id列表，统计出蓝球出现的次数
+        _result = get_analysis_data(_raw_blue_data)  # 根据上步蓝球的出现次数统计，计算偏差值
+        return dict_sort(_result, 'val', True)
 
     elif _method == 'post':  # 当前一期往后num期，不包含此期的数据
-        pass
+        last_blue = _all_data[-1]['blue']  # 获取最后一期蓝球号码
+        blue_id_list = get_blue_id_list_by_blue(_all_data, last_blue)  # 获取所以蓝球等于最后一期蓝球的期数id
+        post_blue_id_list = get_all_post_id_list(_all_data, blue_id_list, _num)
+        _raw_blue_data = get_blue_statistics(post_blue_id_list)  # 根据上步查出的id列表，统计出蓝球出现的次数
+        _result = get_analysis_data(_raw_blue_data)  # 根据上步蓝球的出现次数统计，计算偏差值
+        return dict_sort(_result, 'val', True)
+
     else:
         return False
 
 
-get_blue_forecast(get_all_data(), 'prv', 5)
+if __name__ == "__main__":
+    all_data = get_all_data()
+    history_total = get_delta_blue(get_all_blue(all_data))
+    prv_five = get_blue_forecast(get_all_data(), 'prv', 5)
+    prv_three = get_blue_forecast(get_all_data(), 'prv', 3)
+    post_five = get_blue_forecast(get_all_data(), 'post', 5)
+    post_three = get_blue_forecast(get_all_data(), 'post', 3)
+    print(history_total)
+    print(prv_five)
+    print(prv_three)
+    print(post_five)
+    print(post_three)

orm/db.py

@@ -1,11 +1,13 @@
 from orm_db import History, BlueForecastFiveAll, BlueForecastThreeAll, RedForecastAll
 from sqlalchemy import engine
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy.orm import sessionmaker, declarative_base
 from sqlalchemy import select, desc, asc
 
+# from sqlalchemy import Column, Integer, String, DATE
+
 # 初始化数据库实例
 DB_URI = 'mysql+pymysql://root:Sxzgx1209@home.rogersun.cn:3306/lottery'
-db_engine = engine.create_engine(DB_URI, encoding='utf8', echo=True)
+db_engine = engine.create_engine(DB_URI, echo=False)
 db_session = sessionmaker(bind=db_engine)()
 
 
@@ -39,6 +41,17 @@ def db_get_data_by_date_id(_date_id):
         return False
 
 
+def db_get_data_in_ids(_ids):
+    try:
+        get_in_ids_stmt = select(History).where(History.id.in_(_ids))
+        ids_result = db_session.execute(get_in_ids_stmt).scalars()
+        # print(history_scalars(ids_result))
+        return history_scalars(ids_result)
+    except Exception as e:
+        print(e)
+        return False
+
+
 def history_scalar(_scalar):
     _data = dict()
     _data['id'] = _scalar.id