2.2.1.2-网格交易(python网格交易附实战交易记录)_python实现网格交易

跳转到根目录：知行合一：投资篇

已完成：
1、投资&技术
  1.1.1 投资-编程基础-numpy
  1.1.2 投资-编程基础-pandas
  1.2 金融数据处理
  1.3 金融数据可视化
2、投资方法论
  2.1.1 预期年化收益率
  2.1.2 一个关于y=ax+b的故事
  2.1.3-数据标准化
  2.1.4-相关性分析
  2.2.1.1-一个关于定投的故(姿)事(势)
  2.2.1.2-网格交易
  2.2.1.3-移动平均线

3、投资实证
  [3.1 2023这一年] 被鸽

1. 网格策略说明

网格交易，说最简单的，就是跌了买，涨了卖。

为了少说废话，这里就举1个例子：

1.1. 策略说明

• 网格的大小，可以自己定义，我定义的是0.01作为一个网格

• 买入一笔后，如果后面价格涨了x*0.01（这里x可以自己定，我定的是3），那么就卖出

• 买入一笔后，如果后面价格跌了y*0.01（y一般就是1，也就是1网），那么就继续买入

• 卖出一笔后，如果后面价格跌了x*0.01（这里x可以自己定，我定的是3），那么就买入

• 卖出一笔后，如果后面价格涨了y*0.01（y一般就是1，也就是1网），那么就继续卖出

• 第一天，默认就是买入

• 后续是按照收盘价和之前买入或卖出的价格进行比较，看是否进行买或卖的操作

1.2. 策略数据样例

2016/9/14买入价0.98，那么，我们预期要么是在1.01卖出，要么是在0.97买入

到了2016/9/26，收盘价是0.962，是>0.97的，那我们就直接认为我们挂单了0.97买入

在2016/9/26以0.97买入之后，我们预期要么是在1.00卖出，要么是在0.96买入

到了2016/10/18，收盘价是1.005，是>1.00的，我们可以成交1笔卖出。

2. 策略实现

2.1. 数据结构定义

    def __init__(self, security, start_date=None, end_date=None) -> None:
        super().__init__()
        # security：是这次执行的代码，一般类似512000，510300之类
        # start_date、end_date：会在所有的数据行中，截取对应的时间片段。不传值，那默认就是None，就会用全量数据进行测算。
        # step_price：每个网格的大小，比如我们测算的是512000，是1左右净值的，1%作为网格，是合适的。
        # steps：买入一笔后，如果后面价格涨了x*0.01（这里x可以自己定，我定的是3），那么就卖出。这里的x，就是我们这里的steps
        self.args = {'security': security, 'start_date': start_date, 'end_date': end_date, 'step_price': 0.01, 'steps': 3}
        # 加载数据，后面会具体解释加载过程
        self.daily_df = self.load_data_2_df()
        # 保留本次交易的行数据，方便后面的使用（比如判断上一笔，是买入还是卖出操作）
        self.last_transaction = None
        # 交易历史，最后输出到csv，方便查看
        self.transactions = []
1
2
3
4
5
6
7
8
9
10
11
12
13
'
运行
运行

2.2. 加载数据

    def load_data_2_df(self):
        # 这个案例，是通过既有数据来跑的，512000，包含了从 2016-09-14 ~ 2024-03-29 的日线数据。
        # 如果想要自己获取数据，可以参考之前的文章：https://blog.csdn.net/sdfiiiiii/article/details/135289226，包含了从qstock获取、处理、存储数据等
        df = pd.read_csv("https://gitee.com/kelvin11/public-resources/raw/master/512000.csv")
        # 如果指定了start_date、end_date，就进行数据的切割
        if self.args['start_date']:
            df = df[df["date"] >= self.args['start_date']]
        if self.args['end_date']:
            df = df[df["date"] <= self.args['end_date']]
        # 转换为日期类型
        df['date'] = df['date'].apply(pd.to_datetime, format='%Y-%m-%d')
        # 按照日期的正序排序（防止数据错位）
        df.sort_values(by="date", ascending=True)
        # 设置dataframe的索引，后面取数比较简便一些.
        df = df.set_index("date")
        # 设置了date为索引之后，dataframe里面就没有date这一列了，有时候为了方便处理，还是把date给加上
        df['date'] = df.index.tolist()
        return df
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
'
运行
运行

2.3. 主代码

主代码，意味着，这里是执行流程的核心。

其实也比较简单，就是遍历dataframe，逐行处理数据即可。

    def process(self):
        # dataframe的遍历，逐行处理数据
        for index, row in self.daily_df.iterrows():
            # index是索引，就是日期；row是Series类型，一行完整的数据
            if index != self.daily_df.iloc[0]['date']:
                # 非第一天
                # 1.1 今日收盘价如果 < 上次操作的价格，那么可能要买。
                if row['close'] < self.last_transaction['价格']:
                    # want_buy，为什么可能要买，因为在网格中，如果之前是买入，当天价格下跌没有达到下一网，是不买的。这个逻辑在want_buy内部实现
                    self.want_buy(row=row)
                # 1.2 今日收盘价如果 > 上次操作的价格，那么可能要卖，want_sell
                elif row['close'] > self.last_transaction['价格']:
                    # want_sell其实和want_buy是同样的解释。因为在网格中，如果收盘价大于上次交易价格，但没有达到实际要卖出的价格，那也是不卖的。这个逻辑在want_sell内部实现
                    self.want_sell(row=row)
                else:
                    # hold方法，其实什么都没做，return None。意思就是持有不操作。
                    self.hold(row=row)
                print()
            else:
                # 是第一天，默认就是买入
                # want_buy方法，是指可能要买，为什么可能要买，因为在网格中，如果之前是买入，当天价格下跌没有达到下一网，是不买的。这个逻辑在want_buy内部实现
                self.want_buy(row=row)
        # 遍历完所有的数据之后，将所有的交易记录通过pandas存储到csv文件中
        df = pd.DataFrame(self.transactions)
        df.to_csv('%s交易记录.csv' % self.args['security'], index=None)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
'
运行
运行

2.3.1. 先看看 want_buy

    def want_buy(self, row):
        if row['date'] == self.daily_df.iloc[0]['date']:
            # 这个是第一天的逻辑，直接执行买入。today_buy就真的是买入操作了，主要是记录买入价格和当前行数据
            this_trans = self.today_buy(row, row['close'])
            # 将买入行相关的信息，存储到self.transactions操作历史中，方便整体输出
            self.transactions.append(this_trans)
        else:
            # 这里就不是第一天了，要去判断是否能买的到（比如价格下跌没到下一网位置，就不买的。）
            if self.last_transaction['操作'] == '买':
                # 如果上一次的操作是'买'，那么要构造1个买入价格的阶梯 [上次买入价格, 今日收盘价]，按照'网格大小'构建一个阶梯price_stairs。
                # 用到的是numpy的arange方法，举例:print(np.arange(10,1,-2)) # 输出[10  8  6  4  2]
                start_buy_price = self.last_transaction['价格']
                price_stairs = np.arange(start_buy_price, row['close'], -self.args['step_price'])
                print('前一天是买入，价格%s。收盘价下跌, 构造的买入阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
                if len(price_stairs) >= 1:
                    # 因为，第一个价格是上次买入价格，所以不要包含在本次的买入阶梯里面，做一些切割[1:]
                    price_stairs = price_stairs[1:]
                    for price in price_stairs:
                        # 一般保留3位小数即可
                        price = round(price, 3)
                        # today_buy，是真正的做买入动作了。
                        this_trans = self.today_buy(row, price)
                        # 将买入信息，集中存储到transactions列表中，后续输出到文件
                        self.transactions.append(this_trans)
                else:
                    print('未达到买入阶梯价，今日不执行买入')
            elif self.last_transaction['操作'] == '卖':
                # 构造买入阶梯价，2. 如果"上一次"是卖出，那么要从"-3*网格"开始买
                start_buy_price = self.last_transaction['价格'] - self.args['steps'] * self.args['step_price']
                # 构造价格区间的方法，跟上面是一样的。
                price_stairs = np.arange(start_buy_price, row['close'], -self.args['step_price'])
                print('前一天是卖出，价格%s。收盘价下跌，构造的买入阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
                if start_buy_price == row['close']:
                    # 如果期望开始买的价格，正好是当天收盘价，其实就是此次要买入的价格
                    price_stairs = np.array([start_buy_price])
                if len(price_stairs) >= 1:
                    # 下面的逻辑，跟上面是一样的。
                    for price in price_stairs:
                        price = round(price, 3)
                        this_trans = self.today_buy(row, price)
                        self.transactions.append(this_trans)
                else:
                    print('未达到买入阶梯价，今日不执行买入')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
'
运行
运行

2.3.2. 买入！today_buy

上面方法，用到的today_buy方法，其实就几行代码：

    def today_buy(self, row, price):
        # 扩充行数据元素，增加2列：操作 = 买，价格 = price
        self.daily_df.loc[row['date'], '操作'] = '买'
        self.daily_df.loc[row['date'], '价格'] = price
        # 将买入这个操作的信息，保存在临时变量 self.last_transaction 中，方便后面处理，能快速定位到上次交易是买入还是卖出，以及其价格
        self.last_transaction = self.daily_df.loc[row['date']]
        return self.last_transaction
1
2
3
4
5
6
7
'
运行
运行

2.3.3. want_sell

want_sell和want_buy方法及其相似，就不逐行解释了。

    def want_sell(self, row):
        # 构造卖出阶梯价，1. 如果"前一天"是卖出，那么构造卖出就是按照下一个"网格"卖
        if self.last_transaction['操作'] == '卖':
            start_sell_price = self.last_transaction['价格']
            price_stairs = np.arange(start_sell_price, row['close'], self.args['step_price'])
            print('前一天是卖出，价格%s。收盘价上涨, 构造的卖出阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
            if len(price_stairs) >= 1:
                price_stairs = price_stairs[1:]
                for price in price_stairs:
                    price = round(price, 3)
                    this_trans = self.today_sell(row, price)
                    self.transactions.append(this_trans)
            else:
                print('未达到卖出阶梯价，今日不执行卖出')
        # 构造卖出阶梯价，2. 如果"前一天"是买入出，那么要从"+3*网格"开始卖
        elif self.last_transaction['操作'] == '买':
            start_sell_price = self.last_transaction['价格'] + self.args['steps'] * self.args['step_price']
            price_stairs = np.arange(start_sell_price, row['close'], self.args['step_price'])
            print('前一天是买入，价格%s。收盘价上涨，构造的买卖出阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
            print(price_stairs)
            if start_sell_price == row['close']:
                price_stairs = np.array([start_sell_price])
            if len(price_stairs) >= 1:
                for price in price_stairs:
                    price = round(price, 3)
                    this_trans = self.today_sell(row, price)
                    self.transactions.append(this_trans)
            else:
                print('未达到卖出阶梯价，今日不执行卖出')
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
'
运行
运行

2.3.4. 卖出！today_sell

也是跟today_buy大同小异，没什么要特别说明的地方。

    def today_sell(self, row, price):
        # 扩充行数据元素，增加2列：操作 = 卖，价格 = price
        self.daily_df.loc[row['date'], '操作'] = '卖'
        self.daily_df.loc[row['date'], '价格'] = price
        # 将卖出这个操作的信息，保存在临时变量 self.last_transaction 中，方便后面处理，能快速定位到上次交易是买入还是卖出，以及其价格
        self.last_transaction = self.daily_df.loc[row['date']]
        return self.last_transaction
1
2
3
4
5
6
7
'
运行
运行

2.4. 完整代码

完整的代码已经全部都解释完了，就是上面的几个方法组合起来，就结束，整体代码量，加上充分的注释，163行。

废话少说，上干货，直接就能跑~

import pandas as pd
import numpy as np


class FixedGrid:

    def __init__(self, security, start_date=None, end_date=None) -> None:
        super().__init__()
        # security：是这次执行的代码，一般类似512000，510300之类
        # start_date、end_date：会在所有的数据行中，截取对应的时间片段。不传值，那默认就是None，就会用全量数据进行测算。
        # step_price：每个网格的大小，比如我们测算的是512000，是1左右净值的，1%作为网格，是合适的。
        # steps：买入一笔后，如果后面价格涨了x*0.01（这里x可以自己定，我定的是3），那么就卖出。这里的x，就是我们这里的steps
        self.args = {'security': security, 'start_date': start_date, 'end_date': end_date, 'step_price': 0.01, 'steps': 3}
        # 加载数据，后面会具体解释加载过程
        self.daily_df = self.load_data_2_df()
        # 保留本次交易的行数据，方便后面的使用（比如判断上一笔，是买入还是卖出操作）
        self.last_transaction = None
        # 交易历史，最后输出到csv，方便查看
        self.transactions = []

    def load_data_2_df(self):
        # 这个案例，是通过既有数据来跑的，512000，包含了从 2016-09-14 ~ 2024-03-29 的日线数据。
        # 如果想要自己获取数据，可以参考之前的文章：https://blog.csdn.net/sdfiiiiii/article/details/135289226，包含了从qstock获取、处理、存储数据等
        df = pd.read_csv("https://gitee.com/kelvin11/public-resources/raw/master/512000.csv")
        # 如果指定了start_date、end_date，就进行数据的切割
        if self.args['start_date']:
            df = df[df["date"] >= self.args['start_date']]
        if self.args['end_date']:
            df = df[df["date"] <= self.args['end_date']]
        # 转换为日期类型
        df['date'] = df['date'].apply(pd.to_datetime, format='%Y-%m-%d')
        # 按照日期的正序排序（防止数据错位）
        df.sort_values(by="date", ascending=True)
        # 设置dataframe的索引，后面取数比较简便一些.
        df = df.set_index("date")
        # 设置了date为索引之后，dataframe里面就没有date这一列了，有时候为了方便处理，还是把date给加上
        df['date'] = df.index.tolist()
        return df

    def today_buy(self, row, price):
        # 扩充行数据元素，增加2列：操作 = 买，价格 = price
        self.daily_df.loc[row['date'], '操作'] = '买'
        self.daily_df.loc[row['date'], '价格'] = price
        # 将买入这个操作的信息，保存在临时变量 self.last_transaction 中，方便后面处理，能快速定位到上次交易是买入还是卖出，以及其价格
        self.last_transaction = self.daily_df.loc[row['date']]
        return self.last_transaction

    def today_sell(self, row, price):
        # 扩充行数据元素，增加2列：操作 = 卖，价格 = price
        self.daily_df.loc[row['date'], '操作'] = '卖'
        self.daily_df.loc[row['date'], '价格'] = price
        # 将卖出这个操作的信息，保存在临时变量 self.last_transaction 中，方便后面处理，能快速定位到上次交易是买入还是卖出，以及其价格
        self.last_transaction = self.daily_df.loc[row['date']]
        return self.last_transaction

    def want_buy(self, row):
        if row['date'] == self.daily_df.iloc[0]['date']:
            # 这个是第一天的逻辑，直接执行买入。today_buy就真的是买入操作了，主要是记录买入价格和当前行数据
            this_trans = self.today_buy(row, row['close'])
            # 将买入行相关的信息，存储到self.transactions操作历史中，方便整体输出
            self.transactions.append(this_trans)
        else:
            # 这里就不是第一天了，要去判断是否能买的到（比如价格下跌没到下一网位置，就不买的。）
            if self.last_transaction['操作'] == '买':
                # 如果上一次的操作是'买'，那么要构造1个买入价格的阶梯 [上次买入价格, 今日收盘价]，按照'网格大小'构建一个阶梯price_stairs。
                # 用到的是numpy的arange方法，举例:print(np.arange(10,1,-2)) # 输出[10  8  6  4  2]
                start_buy_price = self.last_transaction['价格']
                price_stairs = np.arange(start_buy_price, row['close'], -self.args['step_price'])
                print('前一天是买入，价格%s。收盘价下跌, 构造的买入阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
                if len(price_stairs) >= 1:
                    # 因为，第一个价格是上次买入价格，所以不要包含在本次的买入阶梯里面，做一些切割[1:]
                    price_stairs = price_stairs[1:]
                    for price in price_stairs:
                        # 一般保留3位小数即可
                        price = round(price, 3)
                        # today_buy，是真正的做买入动作了。
                        this_trans = self.today_buy(row, price)
                        # 将买入信息，集中存储到transactions列表中，后续输出到文件
                        self.transactions.append(this_trans)
                else:
                    print('未达到买入阶梯价，今日不执行买入')
            elif self.last_transaction['操作'] == '卖':
                # 构造买入阶梯价，2. 如果"上一次"是卖出，那么要从"-3*网格"开始买
                start_buy_price = self.last_transaction['价格'] - self.args['steps'] * self.args['step_price']
                # 构造价格区间的方法，跟上面是一样的。
                price_stairs = np.arange(start_buy_price, row['close'], -self.args['step_price'])
                print('前一天是卖出，价格%s。收盘价下跌，构造的买入阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
                if start_buy_price == row['close']:
                    # 如果期望开始买的价格，正好是当天收盘价，其实就是此次要买入的价格
                    price_stairs = np.array([start_buy_price])
                if len(price_stairs) >= 1:
                    # 下面的逻辑，跟上面是一样的。
                    for price in price_stairs:
                        price = round(price, 3)
                        this_trans = self.today_buy(row, price)
                        self.transactions.append(this_trans)
                else:
                    print('未达到买入阶梯价，今日不执行买入')

    def want_sell(self, row):
        # 构造卖出阶梯价，1. 如果"前一天"是卖出，那么构造卖出就是按照下一个"网格"卖
        if self.last_transaction['操作'] == '卖':
            start_sell_price = self.last_transaction['价格']
            price_stairs = np.arange(start_sell_price, row['close'], self.args['step_price'])
            print('前一天是卖出，价格%s。收盘价上涨, 构造的卖出阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
            if len(price_stairs) >= 1:
                price_stairs = price_stairs[1:]
                for price in price_stairs:
                    price = round(price, 3)
                    this_trans = self.today_sell(row, price)
                    self.transactions.append(this_trans)
            else:
                print('未达到卖出阶梯价，今日不执行卖出')
        # 构造卖出阶梯价，2. 如果"前一天"是买入出，那么要从"+3*网格"开始卖
        elif self.last_transaction['操作'] == '买':
            start_sell_price = self.last_transaction['价格'] + self.args['steps'] * self.args['step_price']
            price_stairs = np.arange(start_sell_price, row['close'], self.args['step_price'])
            print('前一天是买入，价格%s。收盘价上涨，构造的买卖出阶梯是：%s' % (self.last_transaction['价格'], price_stairs))
            print(price_stairs)
            if start_sell_price == row['close']:
                price_stairs = np.array([start_sell_price])
            if len(price_stairs) >= 1:
                for price in price_stairs:
                    price = round(price, 3)
                    this_trans = self.today_sell(row, price)
                    self.transactions.append(this_trans)
            else:
                print('未达到卖出阶梯价，今日不执行卖出')

    def hold(self, row):
        return None

    def process(self):
        # dataframe的遍历，逐行处理数据
        for index, row in self.daily_df.iterrows():
            # index是索引，就是日期；row是Series类型，一行完整的数据
            if index != self.daily_df.iloc[0]['date']:
                # 非第一天
                # 1.1 今日收盘价如果 < 上次操作的价格，那么可能要买。
                if row['close'] < self.last_transaction['价格']:
                    # want_buy，为什么可能要买，因为在网格中，如果之前是买入，当天价格下跌没有达到下一网，是不买的。这个逻辑在want_buy内部实现
                    self.want_buy(row=row)
                # 1.2 今日收盘价如果 > 上次操作的价格，那么可能要卖，want_sell
                elif row['close'] > self.last_transaction['价格']:
                    # want_sell其实和want_buy是同样的解释。因为在网格中，如果收盘价大于上次交易价格，但没有达到实际要卖出的价格，那也是不卖的。这个逻辑在want_sell内部实现
                    self.want_sell(row=row)
                else:
                    # hold方法，其实什么都没做，return None。意思就是持有不操作。
                    self.hold(row=row)
                print()
            else:
                # 是第一天，默认就是买入
                # want_buy方法，是指可能要买，为什么可能要买，因为在网格中，如果之前是买入，当天价格下跌没有达到下一网，是不买的。这个逻辑在want_buy内部实现
                self.want_buy(row=row)
        # 遍历完所有的数据之后，将所有的交易记录通过pandas存储到csv文件中
        df = pd.DataFrame(self.transactions)
        df.to_csv('%s交易记录.csv' % self.args['security'], index=None)


if __name__ == '__main__':
    # this = FixedGrid(security='512000', start_date='2020-09-14', end_date='2023-09-14')
    this = FixedGrid(security='512000', start_date=None, end_date=None)
    this.process()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164

3. 收益如何？

其实这里还是要说的，网格这个东西，我们还缺了一个前置条件：

1. 你得有足够的持仓在手里，你才能在想卖的时候卖出，对吧？
2. 你得有足够的money在手机，你才能在想买入的时候买，对吧？

所以，这个策略，是为了告诉我们，如果我都ok，那理论上，一年下来，我能卖出多少回？也就是做了多少次的T，能赚多少钱？

我把这个统计的结果文件，放在了：512000交易记录

为了对这个策略有信心，我只看，每年，卖出了多少回？

插个题外话，这个策略有很多改进的地方，比如：

1. 可以不按照close收盘价来算，而是按照当天的最高或最低价来算
2. 网格，可以有增强，比如，增加网格数量，或者增加买入、卖出份额等

我自己已经非常忠实的执行了网格交易有一段时间，对结果还是比较满意的，贴一下记录，有兴趣的可以交流。