from lsqt.wt import BaseCtaStrategy
from lsqt.wt import CtaContext
import numpy as np
# class MyStrategy 为必填项
class MyStrategy(BaseCtaStrategy):
def __init__(self, name=’pydt_SH510300′, code=’SSE.STK.000002′, barCnt=50, period=’m1′, days=30, k1=0.1, k2=0.1):
BaseCtaStrategy.__init__(self, name)
self.__days__ = days
self.__k1__ = k1
self.__k2__ = k2
self.__period__ = period
self.__bar_cnt__ = barCnt
self.__code__ = code
self.__is_stk__ = True
def on_init(self, context: CtaContext):
code = self.__code__ # 品种代码
context.stra_get_bars(code, self.__period__, self.__bar_cnt__, isMain=True)
context.stra_log_text(“DualThrust inited”)
def on_calculate(self, context: CtaContext):
code = self.__code__ # 品种代码
# 读取最近50条1分钟线(dataframe对象)
df_bars = context.stra_get_bars(code, self.__period__, self.__bar_cnt__, isMain=True)
# 把策略参数读进来，作为临时变量，方便引用
days = self.__days__
k1 = self.__k1__
k2 = self.__k2__
# 平仓价序列、最高价序列、最低价序列
closes = df_bars.closes
highs = df_bars.highs
lows = df_bars.lows
# 读取days天之前到上一个交易日位置的数据
hh = np.amax(highs[-days:-1])
hc = np.amax(closes[-days:-1])
ll = np.amin(lows[-days:-1])
lc = np.amin(closes[-days:-1])
# 读取今天的开盘价、最高价和最低价
# lastBar = df_bars.get_last_bar()
openpx = df_bars.opens[-1]
highpx = df_bars.highs[-1]
lowpx = df_bars.lows[-1]
”’
!!!!!这里是重点
1、首先根据最后一条K线的时间，计算当前的日期
2、根据当前的日期，对日线进行切片,并截取所需条数
3、最后在最终切片内计算所需数据
”’
# 确定上轨和下轨
upper_bound = openpx + k1 * max(hh – lc, hc – ll)
lower_bound = openpx – k2 * max(hh – lc, hc – ll)
# 读取当前仓位
curPos = context.stra_get_position(code)
if curPos == 0:
if highpx >= upper_bound:
context.stra_enter_long(code, 1, ‘enterlong’)
context.stra_log_text(“向上突破%.2f<=%.2f，多仓进场” % (highpx, upper_bound))
return
elif curPos > 0:
if lowpx <= lower_bound:
context.stra_exit_long(code, curPos, ‘exitlong’)
context.stra_log_text(“向下突破%.2f<=%.2f，多仓出场” % (lowpx, lower_bound))
return
def on_tick(self, context: CtaContext, stdCode: str, newTick: dict):
context.stra_log_text(“on tick fired”)
import json
import pandas as pd
config_file_path = ‘config.json’
with open(config_file_path, ‘r’) as file:
config = json.load(file)
data_path = config[‘data_path’]
data = pd.read_csv(data_path)
data = data.head(100)
cci_period = config[‘parameters’][‘cci_period’]
cci_upper_bound = config[‘parameters’][‘cci_upper_bound’]
cci_lower_bound = config[‘parameters’][‘cci_lower_bound’]
“strategy_name”: “CCIStarStrategy”,
“parameters”: {
“code”: “SSE.STK.000002”,
“bar_count”: 50,
“cci_period”: 14,
“cci_upper_bound”: 100,
“cci_lower_bound”: -100
},
“trade_settings”: {
“volume”: 1,
“slippage”: 0.5,
“commission”: 0.1
},
“backtest_settings”: {
“start_date”: “2020-01-01”,
“end_date”: “2021-01-01”
def calculate_cci(high, low, close, period=14):
TP = (high + low + close) / 3
CCI = pd.Series((TP – TP.rolling(window=period).mean()) / (0.015 * TP.rolling(window=period).std()), name=’CCI’)
return CCI
def apply_strategy(data, cci_period=14, cci_upper_bound=100, cci_lower_bound=-100):
data[‘CCI’] = calculate_cci(data[‘High’], data[‘Low’], data[‘Close’], period=cci_period)
data[‘Signal’] = 0
data[‘Signal’][data[‘CCI’] > cci_upper_bound] = 1
data[‘Signal’][data[‘CCI’] < cci_lower_bound] = -1
return data
def plot_signals(data):
plt.figure(figsize=(14, 7))
plt.subplot(2, 1, 1)
plt.plot(data[‘Close’], color=’black’, lw=2.)
plt.plot(data[data[‘Signal’] == 1].index,
data[‘Close’][data[‘Signal’] == 1],
‘^’, markersize=10, color=’g’, lw=0, label=’Buy signal’)
plt.plot(data[data[‘Signal’] == -1].index,
data[‘Close’][data[‘Signal’] == -1],
‘v’, markersize=10, color=’r’, lw=0, label=’Sell signal’)
plt.title(‘Close Price and Signals’)
plt.legend()
plt.subplot(2, 1, 2)
plt.plot(data[‘CCI’], color=’blue’, lw=2.)
plt.axhline(y=cci_upper_bound, color=’g’, linestyle=’–‘)
plt.axhline(y=cci_lower_bound, color=’r’, linestyle=’–‘)
plt.title(‘CCI Indicator’)
plt.show()data = pd.DataFrame({
‘High’: np.random.rand(100) + 100
‘Low’: np.random.rand(100) + 99
‘Close’: np.random.rand(100) + 99.5
})
data_with_signals = apply_strategy(data)
plot_signals(data_with_signal
import matplotlib.pyplot as plt
def plot_strategy(data):
plt.figure(figsize=(14, 7))
plt.plot(data[‘Date’], data[‘Close’], label=’Close Price’)
plt.scatter(data[data[‘Signal’] == 1][‘Date’], data[data[‘Signal’] == 1][‘Close’], label=’Buy Signal’, marker=’^’, color=’green’)
plt.scatter(data[data[‘Signal’] == -1][‘Date’], data[data[‘Signal’] == -1][‘Close’], label=’Sell Signal’, marker=’v’, color=’red’)
plt.legend()
plt.show()plot_strategy(data)