10.矢量化回测

import ta
def lin_reg_trading(symbol):

  def feature_engineering(df):
    """ Create new variables"""

    # We copy the dataframe to avoid interferences in the data
    df_copy = df.dropna().copy()

    # Create the returns
    df_copy['returns'] = df_copy['close'].pct_change(1)

    # Create the SMAs
    df_copy['SMA 15'] = df_copy[['close']].rolling(15).mean().shift(1)
    df_copy['SMA 60'] = df_copy[['close']].rolling(60).mean().shift(1)

    # Create the volatilities
    df_copy['MSD 10'] = df_copy[['returns']].rolling(10).std().shift(1)
    df_copy['MSD 30'] = df_copy[['returns']].rolling(30).std().shift(1)

    # Create the RSI
    RSI = ta.momentum.RSIIndicator(df_copy['close'], window=14, fillna=False)
    df_copy['rsi'] = RSI.rsi()

    return df_copy.dropna()
  
  # Import the data
  df = yf.download(symbol, proxy="http://127.0.0.1:10809")

  # Take adjusted close
  df = df[['Adj Close']]

  # Rename the column
  df.columns = ['close']
  dfc = feature_engineering(df)

  # Percentage train set
  split = int(0.80 * len(dfc))

  # Train set creation
  X_train = dfc[['SMA 15', 'SMA 60', 'MSD 10', 'MSD 30', 'rsi']].iloc[:split]
  Y_train = dfc[['returns']].iloc[:split]

  # Test set creation
  X_test = dfc[['SMA 15', 'SMA 60', 'MSD 10', 'MSD 30', 'rsi']].iloc[split:]
  Y_test = dfc[['returns']].iloc[split:]

  # Import the class
  from sklearn.linear_model import LinearRegression

  # Initialize the calss
  reg = LinearRegression()

  # Fit the model
  reg.fit(X_train, Y_train)

  # Create predictions for the whole dataset
  X = np.concatenate((X_train, X_test), axis=0)
  dfc['prediction'] = reg.predict(X)

  # Compute the position
  dfc['position'] = np.sign(dfc['prediction'])

  # Compute the returns
  dfc['strategy'] = dfc['returns'] * dfc['position'].shift(1)
  dfc['return'] = dfc['strategy']
  BackTest(dfc['return'].iloc[split:])

lin_reg_trading('GC=F')