• このエントリーをはてなブックマークに追加

しばらく放置していた FX システムトレードに新しくストラテジを追加.

Machine Learning for Trading の講義を聞いて、 以下の 2 つが紹介されていたので実装してみた.

スポンサードリンク

線形回帰

statsmodels ライブラリを使う.

class OLS(Strategy):
    def __init__(self, status):
        super(OLS, self).__init__(status)

        self.sma_ols_ts = pd.DataFrame()
        self.mean_for_ols_period = 100
        self.ols_period = 20

        self.a = 0
        self.b = 0
        self.pre_b = 0

    def calc_ols(self, timeseries, event):
        # データをならすために平均をとる.
        mean_for_ols = timeseries.get_latest_ts_as_df(
            self.mean_for_ols_period).mean()[0]
        self.sma_ols_ts.loc[event.time, event.instrument] = mean_for_ols

        x = range(self.ols_period)
        y = np.asarray(
            self.sma_ols_ts.tail(self.ols_period)[event.instrument])

        if len(y) < len(x):
            return

        # いい感じの線分にするために、大きな数をかける
        results = sm.OLS(y * 1000000, sm.add_constant(x), prepend=True).fit()
        self.pre_b = self.b
        self.a, self.b = results.params

        # print(str(self.a) + ", " + str(self.b))

    def ols_buy_condition(self):
        # 傾きが xx 以上のときに buy する
        return self.b > 1.5

    def ols_sell_condition(self):
        # 傾きが-xx 以上のときに sell する
        return self.b < -1.5
    
    def ols_close_buy_condition(self):
        # 傾きがひっくり返ったときに終了
        return self.pre_b >= 0 and self.b < 0

    def ols_close_sell_condition(self):
        # 傾きがひっくり返ったときに終了
        return self.pre_b <= 0 and self.b > 0

    def cleanup_data(self):
        if len(self.sma_ols_ts) > self.ols_period:
            self.sma_ols_ts.drop(self.sma_ols_ts.head(1).index)

K 近傍法

実装してみたものの、バックテストをしてもうまく動いているようには見えない.

なにか、バグっているっぽい.

class KNN(Strategy):
    def __init__(self, status):
        Strategy.__init__(self, status)

        self.knn_train_period = 100
        self.knn_pred_period = 100
        self.knn_neighbors = 10
        self.predict = 0
        self.now = 0
        self.beta = 0
        self.knn_ts = pd.DataFrame()

    def calc_knn(self, timeseries, event):

        x = [[i] for i in range(self.knn_train_period)]
        y = timeseries.get_latest_ts_as_array(self.knn_train_period, event)

        if len(y) < len(x):
            return

        neigh = KNeighborsRegressor(n_neighbors=self.knn_neighbors)
        neigh.fit(x, y)

        self.now = event.bid
        self.predict = neigh.predict(
            self.knn_train_period + self.knn_pred_period)

        if self.status["is_sim"]:
            self.knn_ts.loc[event.time, event.instrument] = self.predict

        self.beta = self.predict - self.now

    def knn_buy_condition(self):
        return self.beta > 0.0001

    def knn_sell_condition(self):
        return self.beta < -0.0001

    def knn_close_buy_condition(self):
        return self.beta <= 0

    def knn_close_sell_condition(self):
        return self.beta >= 0