vn.py官网
开源量化社区
Member
avatar
加入于:
帖子: 4
声望: 0

需求:
选股:每日盘后或开盘前,根据财务指标数据和交易日线数据,按照规则计算排序进行选股,假设选出10支股票。
买入:开盘后根据这10支股票的行情15分钟K线数据,根据规则满足条件买入,单日最多买入3支股票。
卖出:根据持仓的股票情况,以及根据15分钟K线数据,根据规则满足条件则卖出。
问题:
vnpy的多合约组合策略,仍然是策略启动时要设置好固定的合约代码,然后on_init会自动订阅该合约的行情,回测的时候是先根据固定的合约代码加载数据,然后启动策略回放数据,驱动on_bars事件进行回测。
那么在不确定合约的情况下,无法实现该需求。
方案:
首先,采用多合约组合策略为基础,但只订阅一个上证指数合约,作为驱动。
然后在1min的on_bars函数中,判断是否新的一天,如果是则执行一个选股函数,选股函数的数据来源是历史数据库。
然后调用持仓查询函数,获得当前持仓的股票代码。
将选出的10支股票代码加上持仓的股票代码,追加到合约列表,同时调用主引擎去订阅这10支股票。on_tick函数里面合成15mk线数据并驱动on_15m_bars函数。
然后在on_15m_bars里面做买入和卖出策略判断。
感觉这样可以实现实盘,但是回测好像不太好实现。
回测load_data()加载的是上证指数的1mk线数据
然后启动回测,加载策略,调用策略on_init函数,然后驱动on_bar,然后发现是新的1天,调用选股函数,追加到合约列表,到这里都没问题。
接下来,on_tick因为在回测的时候不会触发,也就没法驱动on_bars里面包含新追加的合约的历史数据。
谁有解决办法或者方案吗?

Member
avatar
加入于:
帖子: 4
声望: 0

感觉可以修改回测引擎portfolio_strategy\backtesting.py来实现
修改def run_backtesting(self) 函数

    self.output("开始回放历史数据")

    # Use the rest of history data for running backtesting
    for dt in dts[ix:]:
        try:
            self.new_bars(dt)
        except Exception:
            self.output("触发异常,回测终止")
            self.output(traceback.format_exc())
            return

    self.output("历史数据回放结束")

在self.new_bars(dt)这行之前,取出self.strategy.vt_symbols保存下来,这是on_bars之前的合约代码集合
在self.new_bars(dt)这行之前,再次取出self.strategy.vt_symbols,和之前保存的合约代码集合进行比较
集合相同则不做任何处理,如果不是多股轮动的策略,就走这个分支,逻辑不受影响
集合增加的则对增加的合约执行load_data的逻辑,将新合约的数据添加到history_data当中
集合减少的则对减少的合约执行相反的逻辑,将数据从history_data移除,主要这里担心合约集合不断增加影响效率,其实也可以不做移除

然后修改 def new_bars(self, dt: datetime) 函数
for vt_symbol in self.vt_symbols:
这里要将self.vt_symbols修改成前面的得到的新的合约的集合,这样确保后续self.strategy.on_bars(self.bars)调用的时候self.bars里面包含了最新的合约的行情数据。

这样应该可以吧?哪位大佬帮忙把把脉。

Member
avatar
加入于:
帖子: 2708
声望: 158

可以自己试试看,也可参考https://www.vnpy.com/forum/topic/3180-ru-he-yong-portfolio-strategyshi-xian-shi-zhi-zui-xiao-de-10zhi-gu-piao-yu-qi-huo-dui-chong看看

Member
avatar
加入于:
帖子: 4
声望: 0

xiaohe wrote:

可以自己试试看,也可参考https://www.vnpy.com/forum/topic/3180-ru-he-yong-portfolio-strategyshi-xian-shi-zhi-zui-xiao-de-10zhi-gu-piao-yu-qi-huo-dui-chong看看

感谢回复,那个帖子我看了,思路有相似之处。
多股轮动策略的问题就在于回测加载合约历史数据的不确定性,回测效率肯定会大大降低。
就算15分钟历史数据,假设加载全部3769支股票,能加载多长时间的数据,这应该对内存要求更高,没有测算过。

Member
avatar
加入于:
帖子: 4
声望: 0

经过一段时间研究,基本实现多股轮动策略,成功实现回测,下面是对vnpy引擎代码的一些修改之处:

修改vnpy\app\portfolio_strategy\engine.py
在_init_strategy函数后面增加一个函数:

def add_subscribe(self, strategy: StrategyTemplate, vt_symbols: List[str]):
    # Subscribe market data
    for vt_symbol in vt_symbols:
        contract: ContractData = self.main_engine.get_contract(vt_symbol)
        if contract:
            req = SubscribeRequest(
                symbol=contract.symbol, exchange=contract.exchange)
            self.main_engine.subscribe(req, contract.gateway_name)
        else:
            self.write_log(f"行情订阅失败,找不到合约{vt_symbol}", strategy)


修改vnpy\app\portfolio_strategy\template.py
在第239行增加一个函数:
def add_subscribe(self, vt_symbols: List[str]):
self.strategy_engine.add_subscribe(self, vt_symbols)

修改vnpy\app\portfolio_strategy\backtesting.py
在类PortfolioDailyResult的最后一个函数update_close_prices修改最后追加代码:
原来:
if contract_result:
contract_result.update_close_price(close_price)
修改为:
if contract_result:
contract_result.update_close_price(close_price)
else:
self.contract_results[vt_symbol] = ContractDailyResult(self.date, close_price)

在函数run_backtesting之前增加一个函数:
def add_subscribe(self, strategy: StrategyTemplate, add_vt_symbols: List[str]):
print('追加订阅:', self.vt_symbols)
for vt_symbol in add_vt_symbols:
self.vt_symbols.append(vt_symbol)
print('追加完成:', self.vt_symbols)

    #print('已加载历史时间:',self.dts)
    #print('已加载历史数据:',self.history_data)
    # Load 30 days of data each time and allow for progress update
    progress_delta = timedelta(days=30)
    total_delta = self.end - self.start
    interval_delta = INTERVAL_DELTA_MAP[self.interval]

    for vt_symbol in add_vt_symbols:
        # 追加相应测试参数,从第一个symbol复制一样的参数
        self.size[vt_symbol]=self.size[self.vt_symbols[0]]
        self.rates[vt_symbol]=self.rates[self.vt_symbols[0]]
        self.slippages[vt_symbol]=self.slippages[self.vt_symbols[0]]
        self.priceticks[vt_symbol]=self.priceticks[self.vt_symbols[0]]
        start = self.start
        end = self.start + progress_delta
        progress = 0

        data_count = 0
        while start < self.end:
            end = min(end, self.end)  # Make sure end time stays within set range

            data = load_bar_data(
                vt_symbol,
                self.interval,
                start,
                end
            )

            for bar in data:
                self.dts.add(bar.datetime)
                self.history_data[(bar.datetime, vt_symbol)] = bar
                data_count += 1

            progress += progress_delta / total_delta
            progress = min(progress, 1)
            progress_bar = "#" * int(progress * 10)
            self.output(f"{vt_symbol}加载进度:{progress_bar} [{progress:.0%}]")

            start = end + interval_delta
            end += (progress_delta + interval_delta)

        self.output(f"{vt_symbol}历史数据加载完成,数据量:{data_count}")

    #print('追加加载历史时间:',self.dts)
    #print('追加加载历史数据:',self.history_data)

    self.output("所有历史数据加载完成")



多股轮动策略参照 ,主要如下:
def on_bars(self, bars: Dict[str, BarData]):
里面发现是新的一天,则执行自己的选股函数,得到股票代码,追加 self.vt_symbols.append(onefund.vt_symbol);

        prtlog('追加订阅股票代码:',add_vt_symbols)
        #print(self.pos_cala_datas)

        def on_bar(bar: BarData):
            """"""
            pass

        for vt_symbol in add_vt_symbols:
            # 追加K线管理器
            self.bgs[vt_symbol] = BarGenerator(on_bar)
        # 追加订阅
        self.add_subscribe(add_vt_symbols)
        # TODO:此处要考虑退订,否则策略跑久了,订阅会越来越多,但需要修改引擎,如果速度尚可可不做

另外,如果完美一点,可以继续修改引擎代码,增加退订函数,调用gateway的相应退订接口实现。

© 2015-2019 上海韦纳软件科技有限公司
备案服务号:沪ICP备18006526号-3

沪公网安备 31011502017034号