目录
- 前言
* - T+0限制实现思路
- 一、调整买卖比例并统计pnl
* - 1 – 在main中添加统计pnl
- 2 – 调整买入比例0.98，卖出比例1.02
- 3 – 获取pnl值
- 二、策略添加T+0限制
* - 1 – T+0实现
- 2 – 获取T+0限制后pnl值
- 三、盈亏柱状图对比
* - 1 – 无T+0限制柱状图
- 2 – T+0限制柱状图
- 四、k线图对比
* - 1 – 无T+0限制k线图
- 2 – T+0限制k线图
- 五、完整源码
前言
- 之前我们已经完成了回测，但是我们策略是ma20的单均线策略，这种策略太单一收益也不高
- 我们需要对回归策略的单均线策略进行升级
- 添加T+0交易限制
- 买入需要升级为2次判断，在均线下方做一些累计计算
T+0限制实现思路
- 其实就是在sell方法中添加上日期的判断即可
一、调整买卖比例并统计pnl
1 – 在main中添加统计pnl
if __name__ == '__main__':
orders_df = pd.DataFrame(orders).T
orders_df.loc[:, 'pnl'].plot.bar()
plt.show()
print('sum of pnl is: ' + str(orders_df.loc[:, 'pnl'].sum()))
bar5 = pd.read_csv(bar_path, parse_dates=['datetime'])
bar5.loc[:, 'datetime'] = [date2num(x) for x in bar5.loc[:, 'datetime']]
2 – 调整买入比例0.98，卖出比例1.02
def strategy(self):
if self._is_new_bar:
sum_ = 0
for item in self._Close[1:21]:
sum_ = sum_ + item
self._ma20 = sum_ / 20
if 0 == len(self._current_orders):
if self._Close[0] < 0.98 * self._ma20:
volume = int(100000 / self._Close[0] / 100) * 100
self._buy(self._Close[0] + 0.01, volume)
elif 1 == len(self._current_orders):
if self._Close[0] > self._ma20 * 1.02:
key = list(self._current_orders.keys())[0]
self._sell(key, self._Close[0] - 0.01)
3 – 获取pnl值
; 二、策略添加T+0限制
1 – T+0实现
def strategy(self):
if self._is_new_bar:
sum_ = 0
for item in self._Close[1:21]:
sum_ = sum_ + item
self._ma20 = sum_ / 20
if 0 == len(self._current_orders):
if self._Close[0] < 0.98 * self._ma20:
volume = int(100000 / self._Close[0] / 100) * 100
self._buy(self._Close[0] + 0.01, volume)
elif 1 == len(self._current_orders):
if self._Close[0] > self._ma20 * 1.02:
key = list(self._current_orders.keys())[0]
if self._Dt[0].date() != self._current_orders[key]['open_datetime'].date():
self._sell(key, self._Close[0] - 0.01)
print('open date is %s, close date is: %s.'
% (self._history_orders[key]['open_datetime'].date(), self._Dt[0].date()))
else:
print('sell order aborted due to T+0 limit')
else:
raise ValueError("we have more then 1 current orders")
2 – 获取T+0限制后pnl值
; 三、盈亏柱状图对比
1 – 无T+0限制柱状图
; 2 – T+0限制柱状图
四、k线图对比
1 – 无T+0限制k线图
; 2 – T+0限制k线图
五、完整源码
import requests
from time import sleep
from datetime import datetime, time, timedelta
from dateutil import parser
import pandas as pd
import numpy as np
import os
import matplotlib.pyplot as plt
from mplfinance.original_flavor import candlestick_ohlc
from matplotlib.dates import date2num
def get_ticks_for_backtesting(tick_path, bar_path):
"""
:func: get ticks for backtesting, need two params
:param1 tick_path: 生成的回测数据路径
csv file with tick data,
when there is not tick data,
use bat_path to create tick data
example: "E:\\Downloads\\600036_data\\600036_ticks.csv"
:param2 bar_path: 历史数据的tick路径
csv file with bar data,
used in creating tick data
example: "E:\\Downloads\\600036_data\\600036_5m.csv"
:return: ticks in list with tuples in it, such as
[(datetime, last_price), (datetime, last_price)]
"""
if os.path.exists(tick_path):
ticks = pd.read_csv(
tick_path,
parse_dates=['datetime'],
index_col='datetime'
)
tick_list = []
for index, row in ticks.iterrows():
tick_list.append((index, row[0]))
ticks = tick_list
else:
bar_5m = pd.read_csv(bar_path)
ticks = []
for index, row in bar_5m.iterrows():
if row['open'] < 30:
step = 0.01
elif row['open'] < 60:
step = 0.03
elif row['open'] < 90:
step = 0.05
else:
step = 0.1
arr = np.arange(row['open'], row['high'], step)
arr = np.append(arr, row['high'])
arr = np.append(arr, np.arange(row['open'] - step, row['low'], -step))
arr = np.append(arr, row['low'])
arr = np.append(arr, row['close'])
i = 0
dt = parser.parse(row['datetime']) - timedelta(minutes=5)
for item in arr:
ticks.append((dt + timedelta(seconds=0.1 * i), item))
i += 1
tick_df = pd.DataFrame(ticks, columns=['datetime', 'price'])
tick_df.to_csv(tick_path, index=0)
return ticks
class AstockTrading(object):
"""
:class: A stock trading platform, needs one param,
It has backtesting, paper trading, and real trading.
:param1: strategy_name: strategy_name
"""
def __init__(self, strategy_name):
self._strategy_name = strategy_name
self._Dt = []
self._Open = []
self._High = []
self._Low = []
self._Close = []
self._Volume = []
self._tick = []
self._last_bar_start_minute = None
self._is_new_bar = False
self._ma20 = None
self._current_orders = {}
self._history_orders = {}
self._order_number = 0
self._init = False
def get_tick(self):
"""
:func: for paper trading or real trading, not for backtesting
It goes to sina to get last tick info,
address is: https://hq.sinajs.cn/list=sh600519,
sh600519 can be changed
need to set headers Referer to: https://finance.sina.com.cn
A股的开盘时间是9:15，9:15-9:25是集合竞价 -> 开盘价，9:25
9:25-9:30不交易，时间>9:30，交易开始
start this method after 9:25
tick info is organized in tuple,
such as (trade_datetime, last_price),
tick info is save in self._tick.
:param: no param
:return: None
"""
headers = {'Referer': "https://finance.sina.com.cn"}
page = requests.get("https://hq.sinajs.cn/list=sh600519", headers=headers)
stock_info = page.text
mt_info = stock_info.replace("\"", "").split("=")[1].split(",")
last = float(mt_info[1])
trade_datetime = mt_info[30] + ' ' + mt_info[31]
self._tick = (trade_datetime, last)
def get_history_data_from_local_machine(self):
"""
:not done yet
:return:
"""
self._Open = []
self._High = []
self._Low = []
self._Close = []
self._Dt = []
def bar_generator(self):
"""
:not done yet
:how save and import history data?
:return:
"""
if self._tick[0].minute % 5 == 0 and self._tick[0].minute != self._last_bar_start_minute:
self._last_bar_start_minute = self._tick[0].minute
self._Open.insert(0, self._tick[1])
self._High.insert(0, self._tick[1])
self._Low.insert(0, self._tick[1])
self._Close.insert(0, self._tick[1])
self._Dt.insert(0, self._tick[0])
self._is_new_bar = True
else:
self._High[0] = max(self._High[0], self._tick[1])
self._Low[0] = max(self._Low[0], self._tick[1])
self._Close[0] = self._tick[1]
self._Dt[0] = self._tick[0]
self._is_new_bar = False
def _buy(self, price, volume):
"""
:method: create am order
:param1 price: buying price
:param2 volume: buying volume
:return: none
"""
self._order_number += 1
key = "order" + str(self._order_number)
self._current_orders[key] = {
"open_datetime": self._Dt[0],
"open_price": price,
"volume": volume
}
pass
def _sell(self, key, price):
"""
:method: close a long order, It needs two params
:param1 key: long order's key
:param2 price: selling price
:return:
"""
self._current_orders[key]['close_price'] = price
self._current_orders[key]['close_datetime'] = self._Dt[0]
self._current_orders[key]['pnl'] = \
(price - self._current_orders[key]['open_price']) \
* self._current_orders[key]['volume'] \
- price * self._current_orders[key]['volume'] * 1 / 1000 \
- (price - self._current_orders[key]['open_price']) \
* self._current_orders[key]['volume'] * 3 / 10000
self._history_orders[key] = self._current_orders.pop(key)
def strategy(self):
if self._is_new_bar:
sum_ = 0
for item in self._Close[1:21]:
sum_ = sum_ + item
self._ma20 = sum_ / 20
if 0 == len(self._current_orders):
if self._Close[0] < 0.98 * self._ma20:
volume = int(100000 / self._Close[0] / 100) * 100
self._buy(self._Close[0] + 0.01, volume)
elif 1 == len(self._current_orders):
if self._Close[0] > self._ma20 * 1.02:
key = list(self._current_orders.keys())[0]
if self._Dt[0].date() != self._current_orders[key]['open_datetime'].date():
self._sell(key, self._Close[0] - 0.01)
print('open date is %s, close date is: %s.'
% (self._history_orders[key]['open_datetime'].date(), self._Dt[0].date()))
else:
print('sell order aborted due to T+0 limit')
else:
raise ValueError("we have more then 1 current orders")
def bar_generator_for_backtesting(self, tick):
"""
:method: for backtesting only, used to update _Open, _ High, etc, It needs just one param
:param tick: tick info in tuple, (datetime, price)
:return:
"""
if tick[0].minute % 5 == 0 and tick[0].minute != self._last_bar_start_minute:
self._last_bar_start_minute = tick[0].minute
self._Open.insert(0, tick[1])
self._High.insert(0, tick[1])
self._Low.insert(0, tick[1])
self._Close.insert(0, tick[1])
self._Dt.insert(0, tick[0])
self._is_new_bar = True
else:
self._High[0] = max(self._High[0], tick[1])
self._Low[0] = max(self._Low[0], tick[1])
self._Close[0] = tick[1]
self._Dt[0] = tick[0]
self._is_new_bar = False
def run_backtestting(self, ticks):
"""
:method: ticks will be used to generate bars,
when bars is long enough, call strategy()
:param ticks: list with (datetime, price) in the list
:return: none
"""
for tick in ticks:
self.bar_generator_for_backtesting(tick)
if self._init:
self.strategy()
else:
if len(self._Open) >= 100:
self._init = True
self.strategy()
if __name__ == '__main__':
tick_path = "E:\\Downloads\\600036_data\\600036_ticks.csv"
bar_path = "E:\\Downloads\\600036_data\\600036_5m.csv"
ticks = get_ticks_for_backtesting(tick_path, bar_path)
ast = AstockTrading('ma')
ast.run_backtestting(ticks)
print('ast._current_orders:')
print(ast._current_orders)
print("-------------------------------------")
print('ast._history_orders:')
print(ast._history_orders)
profit_orders = 0
loss_orders = 0
orders = ast._history_orders
for key in orders.keys():
if orders[key]['pnl'] >= 0:
profit_orders += 1
else:
loss_orders += 1
win_rate = profit_orders / len(orders)
loss_rate = loss_orders / len(orders)
orders_df = pd.DataFrame(orders).T
orders_df.loc[:, 'pnl'].plot.bar()
plt.show()
print('sum of pnl is: ' + str(orders_df.loc[:, 'pnl'].sum()))
bar5 = pd.read_csv(bar_path, parse_dates=['datetime'])
bar5.loc[:, 'datetime'] = [date2num(x) for x in bar5.loc[:, 'datetime']]
fig, ax = plt.subplots()
candlestick_ohlc(
ax,
quotes=bar5.values,
width=0.2,
colorup="r",
colordown='g',
alpha=1.0,
)
for index, row in orders_df.iterrows():
ax.plot(
[row['open_datetime'], row['close_datetime']],
[row['open_price'], row['close_price']],
color='darkblue',
marker='o',
)
plt.show()
Original: https://blog.csdn.net/qq23001186/article/details/125359074
Author: 无休止符
Title: 6、Python量化交易-单均线策略升级1：T+0限制
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/752949/
转载文章受原作者版权保护。转载请注明原作者出处！