1

我在带有日期时间索引的 Pandas Dataframe 中有 OHLC 价格数据，并且正在回测交易入场信号。

其中之一是标记买入进入模式，其中先前的显着低点被打破。我在下面编写了可以正常工作的代码，但是如果可能的话，我想对其进行矢量化以返回“FooBuy”列的矢量函数。

我计划最终在实时系统中使用此代码，因此拥有一个自动响应不断变化的数据的向量将是一个很大的好处。

任何人都可以提出任何建议吗？

class FooBuy(PricePattern):

  def df_apply(self, df):

    if not ("SigLow" in df.columns):
       df["SigLow"] = SigLow().df_apply(df)  # correct pattern

    df["SigLowBrokenAt"] = None
    df["FooBuy"] = False

    for date, row in df[df.SigLow].iterrows():
        df_sorted_lows = df[date:].sort("Low", ascending=False)
        df_later_and_lower = df_sorted_lows[df_sorted_lows.Low < row["Low"]]
        if len(df_later_and_lower) > 0:
           df["SigLowBrokenAt"].ix[date] = df_later_and_lower.ix[0].name

    for date in df.SigLowBrokenAt.dropna():
        df["FooBuy"].ix[date] = True

这是 SigLow 的代码：

class PricePattern(object):

  def __init__(self, shift=0, **kwargs):
    self.shift = shift

  def __repr__(self):
    return "{0}({1})".format(self.__class__.__name__, "" if self.shift == 0 else "shift=\"{0}\"".format(self.shift))

  def df_apply(self, df):
    raise NotImplementedError()


class SigLow(PricePattern):

  def __init__(self, **kwargs):
    super(SigLow, self).__init__(**kwargs)

  def df_apply(self, df):
    return (df.Low.shift(self.shift) < df.Low.shift(self.shift + 1)) & \
        (df.Low.shift(self.shift) < df.Low.shift(self.shift - 1))

这是一些通过的测试代码：

class TestBase(unittest.TestCase):
  def _create_df(self, data):
    return 1 + pd.DataFrame(data, columns=['Open', 'High', 'Low', 'Close'],
                            index=pd.date_range('2/25/2013 06:00', periods=len(data),
                                                freq='H')) / 100

  def _assert_true_alone(self, series, index):
    for i in arange(len(series)):
        self.assertTrue(series[i] or (i not in index))

  def _assert_all_false(self, series):
    for i in arange(len(series)):
        self.assertFalse(series[i])


class TestFooBuy(TestBase):

  def test_matches(self):
    data1 = [[48, 49, 34, 48],
             [48, 50, 46, 48],
             [47, 49, 38, 48],  # SL
             [48, 59, 48, 58],
             [57, 57, 50, 54],
             [53, 53, 36, 40],  # FB
             [39, 39, 29, 32]]

    df = self._create_df(data1)
    #df["FooBuy"] = FooBuy().df_apply(df)  # correct pattern
    FooBuy().df_apply(df)
    self._assert_true_alone(df.SigLow, [2])
    self._assert_true_alone(df.FooBuy, [5])

  def test_does_not_match(self):
    data1 = [[48, 49, 34, 48],
             [48, 50, 46, 48],
             [47, 49, 38, 48],  # SL
             [48, 59, 48, 58],
             [57, 57, 42, 48],  # SL
             [48, 50, 43, 44],
             [45, 52, 45, 51]]

    df = self._create_df(data1)
    #df["FooBuy"] = FooBuy().df_apply(df, config_defaults)  # correct pattern
    FooBuy().df_apply(df, config_defaults)
    #print df
    self._assert_true_alone(df.SigLow, [2, 4])
    self._assert_all_false(df.FooBuy)
4

0 回答 0