Lottery/history_result.py

import pymysql
from functools import reduce
import json
from pymysql.cursors import DictCursor
from blue_forecast import dict_sort

db = {
    'host': 'home.rogersun.online',
    'user': 'root',
    'password': 'Sxzgx1209',
    'database': 'lottery'
}


def dict_rate(_dict):
    _sum = reduce((lambda x, y: x + y), _dict.values())
    for k, v in _dict.items():
        _dict[k] = str(round(v / _sum * 100, 2)) + '%'
    return _dict


def get_blue_rate(_date_id, _range):
    conn = pymysql.connect(**db)
    cursor = conn.cursor(DictCursor)

    # 获取指定一期的id和篮球号码
    cursor.execute('SELECT id, blue FROM history WHERE dateId = %s', (_date_id,))
    d = cursor.fetchall()[0]
    data_id = d['id']
    blue = d['blue']

    # 获取指定一期之前的全部篮球相同的数据
    cursor.execute('SELECT * FROM history WHERE id <= %s AND blue = %s', (data_id, blue))
    history_data = cursor.fetchall()

    id_list = []
    for data in history_data:
        for r in range(_range):
            id_list.append(data['id'] + r + 1)
    id_list = set(id_list)

    cursor.execute('SELECT * FROM history WHERE id IN %(ids)s', {'ids': id_list})
    result = cursor.fetchall()

    cursor.close()
    conn.commit()

    result_dict = {}

    for r in result:
        b = r['blue']
        if b in result_dict.keys():
            result_dict[b] += 1
        else:
            result_dict[b] = 1
    return dict_sort(result_dict, 'val', True)


def get_result(_date_id, _range):
    conn = pymysql.connect(**db)
    cursor = conn.cursor(DictCursor)

    # 获取指定一期的id和篮球号码
    cursor.execute('SELECT id, blue FROM history WHERE dateId = %s', (_date_id,))
    d = cursor.fetchall()[0]
    data_id = d['id']
    data_blue = d['blue']

    cursor.execute('SELECT * FROM history WHERE id = %s', (data_id + 1,))
    result = cursor.fetchall()[0]
    result_blue = result['blue']

    blue_list = get_blue_rate(_date_id, _range)

    index = [blue[0] for blue in blue_list.items()].index(result_blue) + 1

    print(
        f"{_date_id}: 篮球为{data_blue} 预测下期{json.dumps(dict_rate(blue_list))} 下期篮球实际为{result_blue} 预测索引位置 {index}")
    return index


def get_all_history_result(_date_id, _range):
    index_rate = {}
    conn = pymysql.connect(**db)
    cursor = conn.cursor(DictCursor)

    # 获取指定一期的id和篮球号码
    cursor.execute('SELECT id FROM history WHERE dateId = %s', (_date_id,))
    data_id = cursor.fetchall()[0]['id']
    if data_id > 1000:
        cursor.execute('SELECT dateId FROM history WHERE id <= %s AND id > 1000', (data_id,))
        all_date_id_list = cursor.fetchall()
        print(len(all_date_id_list))

        for _id in all_date_id_list:
            _index = str(get_result(_id['dateId'], _range))
            if _index in index_rate.keys():
                index_rate[_index] += 1
            else:
                index_rate[_index] = 1
    else:
        print('采样数据不足')

    index_rate_result = dict_sort(index_rate, 'val', True)
    print('篮球出现在每次预测的索引位置规律：')
    print(json.dumps(dict_rate(index_rate_result)))


if __name__ == "__main__":
    # get_result('2006136', 1)
    get_all_history_result('2019101', 1)