代码分析和解释

代码是通过线性回归分析,获取指定股票、指数或基金在过去特定年份内的收盘价的线性回归期望值、残差标准差等统计参数,并将结果保存到 SQLite 数据库中。

主要的操作流程包括:

  1. 定义 linear_regression_stock_multi 函数,用于对单一股票、指数或基金进行线性回归分析,并返回包含线性回归统计参数的 DataFrame。

  2. 定义 get_circulate_xslx_str 函数,用于读取 Excel 文件中的股票代码列表,循环调用 linear_regression_stock_multi 函数,将所有股票的线性回归分析结果合并成一个大的 DataFrame,并保存到 SQLite 数据库中。

  3. 在 if __name__ == '__main__': 代码块中,调用 get_circulate_xslx_str 函数,将指定 Excel 文件中的所有股票代码的线性回归分析结果保存到数据库中,并打印输出。

此代码利用了 akshare, pandas, numpy, scipy, sqlite3 和 datetime 等库的功能。

打开网易新闻 查看更多图片

文章代码块(md 格式):

import akshare as ak import pandas as pd import numpy as np import sqlite3 from scipy import stats from datetime import datetime, timedelta def linear_regression_stock_multi(symbol, kind, years_list): df_list = [] for many_years in years_list: today = datetime.now().date() start_date = (today - timedelta(days=365*many_years)).strftime('%Y%m%d') end_date = today.strftime('%Y%m%d') if kind == '指数': df = ak.index_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date) elif kind == '股票': df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") elif kind == '基金': df = ak.fund_etf_hist_em(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") y = df["收盘"] x = np.arange(len(y)) slope, intercept, r_value, p_value, std_err = stats.linregress(x, y) expected_value = intercept + slope * len(y) residuals = y - (intercept + slope * x) std_residuals = np.std(residuals) index=[f"{many_years}year_expected_value", f"{many_years}year_std_residuals", f"{many_years}year_slope", f"{many_years}year_intercept", f"{many_years}year_r_value", f"{many_years}year_p_value", f"{many_years}year_std_err"] data = [expected_value, std_residuals, slope, intercept, r_value, p_value, std_err] result_df = pd.DataFrame(data=[data], index=[symbol], columns=index) df_list.append(result_df) result = pd.concat(df_list, axis=1) return result def get_circulate_xslx_str(kind, file_index_code, sheet): codes = pd.read_excel(file_index_code, sheet_name=sheet, engine='openpyxl', dtype={"代码": str})['代码'].tolist() all_data = pd.DataFrame() for code in codes: try: ratios = linear_regression_stock_multi(code, kind, [7,3,1]) all_data = pd.concat([all_data, ratios]) except KeyError as e: print(f"无法获取股票代码 {code} 的数据: {e}") conn = sqlite3.connect("D:\\wenjian\\python\\smart\\data\\req_data.db") all_data.to_sql("股票", conn, if_exists="replace", index=False) conn.close() return all_data if __name__ == '__main__': print(get_circulate_xslx_str('股票', '测试.xlsx', 0))