股价波动趋势预测demo-基于Tushare数据库（股市行情最新预测）

import pandas as pd import tushare as ts from datetime import timedelta, datetime def get_origin_data(ts_code="000001.SH", past_day="", token=""): """获取数据 :param ts_code: 股票代码，默认上证综指 :param past_day: 获取今天之前多少天的数据，用于确定数据开始日期，默认是2000.1.1开始 :return: pandas.Dataframe tushare包返回的数据 """ ts.set_token(token) pro = ts.pro_api() today_date_str = datetime.strftime(datetime.now(), "%Y%m%d") if past_day: past_date_str = datetime.strftime(datetime.now() - timedelta(days=past_day), "%Y%m%d") else: # start from 2000.1.1 past_date_str = 20000101 print(today_date_str, past_date_str) if ts_code == "000001.SH": data = pro.index_daily(ts_code=ts_code, start_date=past_date_str, end_date=today_date_str) else: data = pro.daily(ts_code=ts_code, start_date=past_date_str, end_date=today_date_str) data.loc[:, is_up] = data.change.map(lambda x: int(x >= 0)).shift(-1) data.loc[:, trade_date] = pd.to_datetime(data[trade_date].values) return data

def gen_cv_data_dict(data, train_set_len, test_set_len, feature_cols, target_col="is_up"): """构建训练集和测试集 :param data: 获取了特征之后的数据 pandas.Dataframe :param train_set_len: 训练集的长度 int. :param test_set_len: 测试集的长度 int. :return: """ cv_data_dict = {} total_data_num = data.shape[0] set_num = int(np.ceil((total_data_num - train_set_len) / test_set_len)) # 正着取max_len，数据是从倒序排的 all_trade_date_list = [datetime.strptime(datetime.strftime(day, "%Y%m%d"), "%Y%m%d") for day in data[trade_date].tolist()] print(all_trade_date_list[:10], type(all_trade_date_list[0]), all_trade_date_list[0]) # 第一个是最近的时间 this_week_monday = get_this_monday(datetime.now()) # print(f"this_week_start:{this_week_monday}", type(this_week_monday)) before_date = get_trade_day_before(anchor_date=this_week_monday) print(f"this_week_monday:{this_week_monday}, before_date: {before_date}") before_date_index = all_trade_date_list.index(before_date) print(f"before_date_index:{before_date_index}") for i in range(set_num): if i == 0: test_start_index = i + before_date_index else: test_start_index = (i + before_date_index) * test_set_len test_end_index = test_start_index + test_set_len train_end_index = test_end_index + 30 + train_set_len train_data = data.iloc[test_end_index + 30:train_end_index, :] test_data = data.iloc[test_start_index:test_end_index, :] print(test_data) if not test_data.empty: test_start_date = datetime.strftime(test_data[trade_date].tolist()[-1], "%Y%m%d") test_end_date = datetime.strftime(test_data[trade_date].tolist()[0], "%Y%m%d") print( # train_data[trade_date].to_list()[0], train_data[trade_date].to_list()[-1], # test_data[trade_date].to_list()[0], test_data[trade_date].to_list()[-1], f"{test_start_date} - {test_end_date}" ) # if train_data.shape[0] <= cv_data_dict[i] = {X_train: train_data[feature_cols], X_test: test_data[feature_cols], y_train: train_data[target_col].values, y_test: test_data[target_col].values, test_date: {}_{}.format(test_start_date, test_end_date), } # print("第{}组：{}-{}，训练集数据量：{}，测试集数据量：{}，日期：{}".format( # i, start_index, end_index, train_data.shape, test_data.shape, datetime.strftime(test_data[trade_date].tolist()[0], "%Y%m%d") # )) return cv_data_dict

from sklearn.ensemble import RandomForestClassifier def main_runner(cleaned_data, used_columns, train_set_len, test_set_len): cv_data_dict = gen_cv_data_dict(data=cleaned_data, train_set_len=train_set_len, test_set_len=test_set_len, feature_cols=used_columns,) for key, _data in cv_data_dict.items(): # 第一次训练和之后训练 print("第{key}组数据：训练集数据量：{train_set}，测试集数据量：{test_set}，模型名称：{model_name}，模型日期：{model_date}".format( key=key, train_set=_data[X_train].shape[0], test_set=_data[X_test].shape[0], model_name=随机森林, model_date=_data[test_date], )) if _data[X_train].shape[0] >= 100: RandomForestClassifier(_data) else: print("没数据啦")