Datawhale-quant 笔记03 - 量化选股策略浅尝试

import baostock as bs
import pandas as pd
import matplotlib.pyplot as plt
import statsmodels.api as sm

from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split

def get_bs_df(code, list_para, start_date, end_date, frequency='d', adjustflag='3'):
    para = ','.join(list_para)
    rs = bs.query_history_k_data_plus(code,para,
    start_date=start_date, end_date=end_date,
    frequency=frequency, adjustflag=adjustflag)
    data_list = []
    while (rs.error_code == '0') & rs.next():
        data_list.append(rs.get_row_data())
    df_result = pd.DataFrame(data_list, columns=rs.fields)
    print(f'get {len(df_result)} records!')
    return df_result
    

def get_bs_profit_data(code, year, quarter):
    profit_list = []
    rs_profit = bs.query_profit_data(code=code, year=year, quarter=quarter)
    while (rs_profit.error_code == '0') & rs_profit.next():
        profit_list.append(rs_profit.get_row_data())
    df_result = pd.DataFrame(profit_list, columns=rs_profit.fields)
    return df_result

lg = bs.login()

code = 'sh.600000'
list_para = ['date', 'code', 'close', 'volume', 'turn', 'peTTM', 'psTTM']
start_date = '2020-01-01'
end_date = '2024-01-20'

df_daily = get_bs_df(code, list_para, start_date, end_date)
df_profit = get_bs_profit_data(code, '2020', '1')

df_profit

# get market value and earning
df_daily['market_value'] = df_daily['close'].astype(float) * float(df_profit['liqaShare'].iloc[0])
df_daily['peTTM'] = df_daily['peTTM'].replace('', pd.NaT)
df_daily['peTTM'] = df_daily['peTTM'].astype(float)
df_daily['peTTM'] = df_daily['peTTM'].fillna((df_daily['peTTM'].shift() + df_daily['peTTM'].shift(-1)) / 2)
df_daily['earning'] = df_daily['market_value'] / df_daily['peTTM'].astype(float)
df_daily

df_daily['market_value'].plot()

df_daily['earning'].plot()

groups = df_daily.groupby('date')
groups

# 定义一个函数来执行市值中性化
def market_neutralize(group):
    # 提取市值和收益的数据列
    market_cap = group['market_value']
    returns = group['earning']
    # 添加截距项
    X = sm.add_constant(market_cap)
    # 执行线性回归，拟合收益率与市值的关系
    model = sm.OLS(returns, X)
    results = model.fit()
    # 提取回归系数
    beta = results.params['market_value']
    # 计算市值中性化后的收益
    neutralized_returns = returns - beta * market_cap
    # 将市值中性化后的收益添加到数据框中
    group['neutralized_value'] = neutralized_returns
    return group

neutralized_data = groups.apply(market_neutralize)

neutralized_data

neutralized_data['neutralized_value'].plot()

def get_all_stock(day):
    rs = bs.query_all_stock(day=day)
    print('login respond  error_msg:'+lg.error_msg)
    data_list = []
    while (rs.error_code == '0') & rs.next():
        data_list.append(rs.get_row_data())
    df_result = pd.DataFrame(data_list, columns=rs.fields)
    return df_result

df_all_stock = get_all_stock('2023-06-30')
df_all_stock

df_all_stock

random_code_list = df_all_stock['code'].sample(n=100, random_state=42)
random_code_list[:10]

df_random_stock = pd.DataFrame()
for code in random_code_list:
    df_query_profit = get_bs_profit_data(code, '2023', '1')
    df_random_stock = pd.concat([df_query_profit, df_random_stock])
df_random_stock

for col in ['roeAvg', 'npMargin', 'gpMargin', 'epsTTM']:
    df_random_stock[col] = df_random_stock[col].replace('', float('nan'))
    df_random_stock[col] = df_random_stock[col].astype(float)
    df_random_stock[col] = df_random_stock[col].fillna((df_random_stock[col].shift() + df_random_stock[col].shift(-1)) / 2)
    df_random_stock[col] = df_random_stock[col].ffill().bfill()

df_random_stock.info()

X_test

X = df_random_stock[['roeAvg', 'npMargin', 'gpMargin']] # 特征因子
y = df_random_stock['epsTTM'] # 目标变量

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
scaler = StandardScaler()
X_train = scaler.fit_transform(X_train)
X_test = pd.DataFrame(scaler.transform(X_test))
model = LinearRegression()
model.fit(X_train, y_train)
print('Factor weights:', model.coef_)
y_pred = model.predict(X_test)
predicted_returns = pd.DataFrame({
    'Stock': X_test.index,
    'Predicted return': y_pred
})
selected_stocks = predicted_returns[predicted_returns['Predicted return'] > 0.1]
print('Selected stocks:', selected_stocks)

y_pred

pd.Series(y_test)

pd.DataFrame({'y_pred':y_pred, 'y_test':y_test, 'diff %':(y_pred-y_test)/y_test*100})