金融小课堂 | 零基础30天API量化速成_第14讲

?“量化学习之算法篇”

即使你并无代码的经验，但只要您学会如何在Quantopian平台上克隆这些极为有利可图的算法代码，多多练习回测和交易，就能为您带来不小的收获。

以下算法来自世界各地的开放作者社区提交，资金分配给了八个国家的作者，其中包括澳大利亚，加拿大，中国，哥伦比亚，印度，西班牙和美国。

这八个算法均在Medium上公布，它们分别是：

Zack’s Long-Short PEAD with News Sentiment and the Street’s Consensus (LIVE TRADING)

Zack’s Long PEAD with News Sentiment

Are Earnings Predictable with Buyback Announcements? (LIVE TRADING)

Reversals During Earnings Announcements (LIVE TRADING)

Clenow Momentum Strategy (as Amended)

VIX Robinhood Momentum Strategy (LIVE TRADING)

JAVoIS: Just Another Volatility Strategy (LIVE TRADING)

101 Alphas Project: Alpha# 41

它们都有几个共同点：

1）展示出持续盈利的回测；

2）使用广泛的股票并广泛分配资本，而与任何特定的股票或行业无关；

3）与市场无关；

4）符合Quantopian团队规定的标准；

要知道，选用不同的量化交易算法所带来的组合收益截然不同。上一篇中我们讲解了ALGO-1 Zack’s Long-Short PEAD：金融小课堂 | 零基础30天API量化速成_第13讲，接下来介绍第二种算法：

ALGO - 2

Zack’s Long PEAD with News Sentiment

收益公告期是每种股票生命中的特殊时期。股票受到了越来越多的审查，投资者和交易者对与它们有关的所有新闻都做出了更加积极的反应。

ALGO-2Zack’s Long PEAD with News Sentiment算法的目的是对冲公告发布后的价格浮动，与前一篇文章中讲解的Zack’s Long-Short PEAD一样，Long PEAD也使用了Zack和Accern的数据。但是，当预期浮动为正时，该算法仅持有多头头寸。

完整代码如下（来源github）：

import numpy as np?from quantopian.algorithm import attach_pipeline, pipeline_outputfrom quantopian.pipeline import Pipelinefrom quantopian.pipeline.data.builtin import USEquityPricingfrom quantopian.pipeline.factors import CustomFactor, AverageDollarVolumefrom quantopian.pipeline.filters.morningstar import Q500US, Q1500USfrom quantopian.pipeline.data import morningstar as mstarfrom quantopian.pipeline.classifiers.morningstar import Sectorfrom quantopian.pipeline.filters.morningstar import IsPrimaryShare?from quantopian.pipeline.data.zacks import EarningsSurprisesfrom quantopian.pipeline.factors.zacks import BusinessDaysSinceEarningsSurprisesAnnouncement??# from quantopian.pipeline.data.accern import alphaone_free as alphaone# Premium version availabe at# https://www.quantopian.com/data/accern/alphaonefrom quantopian.pipeline.data.accern import alphaone as alphaone?def make_pipeline(context): # Create our pipeline pipe = Pipeline() ? # Instantiating our factors factor = EarningsSurprises.eps_pct_diff_surp.latest? # Filter down to stocks in the top/bottom according to # the earnings surprise longs = (factor >= context.min_surprise) & (factor <= context.max_surprise) #shorts = (factor <= -context.min_surprise) & (factor >= -context.max_surprise) ''' change value of q_filters to Q1500US to use Q1500US universe ''' q_filters = Q500US #q_filter1 = Q1500US? # Set our pipeline screens # Filter down stocks using sentiment article_sentiment = alphaone.article_sentiment.latest top_universe = q_filters() & universe_filters() & longs & article_sentiment.notnan() \ & (article_sentiment > .30) # bottom_universe = q_filters() & universe_filters() & shorts & article_sentiment.notnan() \& (article_sentiment < -.30)? # Add long/shorts to the pipeline pipe.add(top_universe, "longs") # pipe.add(bottom_universe, "shorts") pipe.add(BusinessDaysSinceEarningsSurprisesAnnouncement(), 'pe') pipe.set_screen(factor.notnan()) return pipe def initialize(context): #: Set commissions and slippage to 0 to determine pure alpha ''' set_commission(commission.PerShare(cost=0, min_trade_cost=0)) set_slippage(slippage.FixedSlippage(spread=0)) set_slippage(slippage.FixedSlippage(spread=0.02)) set_commission(commission.PerTrade(cost=5.00)) set_slippage(TradeAtTheOpenSlippageModel(0.2,.05)) set_commission(commission.PerShare(cost=0.01)) ''' #: Declaring the days to hold, change this to what you want context.days_to_hold = 3 #: Declares which stocks we currently held and how many days we've held them dict[stock:days_held] context.stocks_held = {}? #: Declares the minimum magnitude of percent surprise context.min_surprise = .00 context.max_surprise = .05? #: OPTIONAL - Initialize our Hedge # See order_positions for hedging logic # context.spy = sid(8554) # Make our pipeline attach_pipeline(make_pipeline(context), 'earnings')? # Log our positions at 10:00AM schedule_function(func=log_positions, date_rule=date_rules.every_day(), time_rule=time_rules.market_close(minutes=30)) # Order our positions schedule_function(func=order_positions, date_rule=date_rules.every_day(), time_rule=time_rules.market_open())?def before_trading_start(context, data): # Screen for securities that only have an earnings release # 1 business day previous and separate out the earnings surprises into # positive and negative results = pipeline_output('earnings') results = results[results['pe'] == 1] assets_in_universe = results.index context.positive_surprise = assets_in_universe[results.longs] #context.negative_surprise = assets_in_universe[results.shorts]?def log_positions(context, data): #: Get all positions if len(context.portfolio.positions) > 0: all_positions = "Current positions for %s : " % (str(get_datetime())) for pos in context.portfolio.positions: if context.portfolio.positions[pos].amount != 0: all_positions += "%s at %s shares, " % (pos.symbol, context.portfolio.positions[pos].amount) log.info(all_positions) def order_positions(context, data): """ Main ordering conditions to always order an equal percentage in each position so it does a rolling rebalance by looking at the stocks to order today and the stocks we currently hold in our portfolio. """ port = context.portfolio.positions record(leverage=context.account.leverage)? # Check our positions for loss or profit and exit if necessary check_positions_for_loss_or_profit(context, data) # Check if we've exited our positions and if we haven't, exit the remaining securities # that we have left for security in port: if data.can_trade(security): if context.stocks_held.get(security) is not None: context.stocks_held[security] += 1 if context.stocks_held[security] >= context.days_to_hold: order_target_percent(security, 0) del context.stocks_held[security] # If we've deleted it but it still hasn't been exited. Try exiting again else: log.info("Haven't yet exited %s, ordering again" % security.symbol) order_target_percent(security, 0) ? # Check our current positions current_positive_pos = [pos for pos in port if (port[pos].amount > 0 and pos in context.stocks_held)] #current_negative_pos = [pos for pos in port if (port[pos].amount < 0 and pos in context.stocks_held)] #negative_stocks = context.negative_surprise.tolist() + current_negative_pos positive_stocks = context.positive_surprise.tolist() + current_positive_pos ''' # Rebalance our negative surprise securities (existing + new) for security in negative_stocks: can_trade = context.stocks_held.get(security) <= context.days_to_hold or \ context.stocks_held.get(security) is None if data.can_trade(security) and can_trade: order_target_percent(security, -1.0 / len(negative_stocks)) if context.stocks_held.get(security) is None: context.stocks_held[security] = 0 ''' # Rebalance our positive surprise securities (existing + new) for security in positive_stocks: can_trade = context.stocks_held.get(security) <= context.days_to_hold or \ context.stocks_held.get(security) is None if data.can_trade(security) and can_trade: order_target_percent(security, 1.0 / len(positive_stocks)) if context.stocks_held.get(security) is None: context.stocks_held[security] = 0? #: Get the total amount ordered for the day # amount_ordered = 0 # for order in get_open_orders(): # for oo in get_open_orders()[order]: # amount_ordered += oo.amount * data.current(oo.sid, 'price')? #: Order our hedge # order_target_value(context.spy, -amount_ordered) # context.stocks_held[context.spy] = 0 # log.info("We currently have a net order of $%0.2f and will hedge with SPY by ordering $%0.2f" % (amount_ordered, -amount_ordered)) def check_positions_for_loss_or_profit(context, data): # Sell our positions on longs/shorts for profit or loss for security in context.portfolio.positions: is_stock_held = context.stocks_held.get(security) >= 0 if data.can_trade(security) and is_stock_held and not get_open_orders(security): current_position = context.portfolio.positions[security].amount cost_basis = context.portfolio.positions[security].cost_basis price = data.current(security, 'price') # On Long & Profit if price >= cost_basis * 1.10 and current_position > 0: order_target_percent(security, 0) log.info( str(security) + ' Sold Long for Profit') del context.stocks_held[security] ''' # On Short & Profit if price <= cost_basis* 0.90 and current_position < 0: order_target_percent(security, 0) log.info( str(security) + ' Sold Short for Profit') del context.stocks_held[security] ''' # On Long & Loss if price <= cost_basis * 0.90 and current_position > 0: order_target_percent(security, 0) log.info( str(security) + ' Sold Long for Loss') del context.stocks_held[security] ''' # On Short & Loss if price >= cost_basis * 1.10 and current_position < 0: order_target_percent(security, 0) log.info( str(security) + ' Sold Short for Loss') del context.stocks_held[security] '''# Constants that need to be globalCOMMON_STOCK= 'ST00000001'?SECTOR_NAMES = { 101: 'Basic Materials', 102: 'Consumer Cyclical', 103: 'Financial Services', 104: 'Real Estate', 205: 'Consumer Defensive', 206: 'Healthcare', 207: 'Utilities', 308: 'Communication Services', 309: 'Energy', 310: 'Industrials', 311: 'Technology' ,}?# Average Dollar Volume without nanmean, so that recent IPOs are truly removedclass ADV_adj(CustomFactor): inputs = [USEquityPricing.close, USEquityPricing.volume] window_length = 252 def compute(self, today, assets, out, close, volume): close[np.isnan(close)] = 0 out[:] = np.mean(close * volume, 0)def universe_filters(): # Equities with an average daily volume greater than 750000. high_volume = (AverageDollarVolume(window_length=252) > 750000) # Not Misc. sector: sector_check = Sector().notnull() # Equities that morningstar lists as primary shares. # NOTE: This will return False for stocks not in the morningstar database. primary_share = IsPrimaryShare() # Equities for which morningstar's most recent Market Cap value is above $300m. have_market_cap = mstar.valuation.market_cap.latest > 300000000 # Equities not listed as depositary receipts by morningstar. # Note the inversion operator, `~`, at the start of the expression. not_depositary = ~mstar.share_class_reference.is_depositary_receipt.latest # Equities that listed as common stock (as opposed to, say, preferred stock). # This is our first string column. The .eq method used here produces a Filter returning # True for all asset/date pairs where security_type produced a value of 'ST00000001'. common_stock = mstar.share_class_reference.security_type.latest.eq(COMMON_STOCK) # Equities whose exchange id does not start with OTC (Over The Counter). # startswith() is a new method available only on string-dtype Classifiers. # It returns a Filter. not_otc = ~mstar.share_class_reference.exchange_id.latest.startswith('OTC') # Equities whose symbol (according to morningstar) ends with .WI # This generally indicates a "When Issued" offering. # endswith() works similarly to startswith(). not_wi = ~mstar.share_class_reference.symbol.latest.endswith('.WI') # Equities whose company name ends with 'LP' or a similar string. # The .matches() method uses the standard library `re` module to match # against a regular expression. not_lp_name = ~mstar.company_reference.standard_name.latest.matches('.* L[\\. ]?P\.?$') # Equities with a null entry for the balance_sheet.limited_partnership field. # This is an alternative way of checking for LPs. not_lp_balance_sheet = mstar.balance_sheet.limited_partnership.latest.isnull() # Highly liquid assets only. Also eliminates IPOs in the past 12 months # Use new average dollar volume so that unrecorded days are given value 0 # and not skipped over # S&P Criterion liquid = ADV_adj() > 250000 # Add logic when global markets supported # S&P Criterion domicile = True # Keep it to liquid securities ranked_liquid = ADV_adj().rank(ascending=False) < 1500 universe_filter = (high_volume & primary_share & have_market_cap & not_depositary & common_stock & not_otc & not_wi & not_lp_name & not_lp_balance_sheet & liquid & domicile & sector_check & liquid & ranked_liquid) return universe_filter# Slippage model to trade at the open or at a fraction of the open - close range. class TradeAtTheOpenSlippageModel(slippage.SlippageModel): '''Class for slippage model to allow trading at the open or at a fraction of the open to close range. ''' # Constructor, self and fraction of the open to close range to add (subtract) # from the open to model executions more optimistically def __init__(self, fractionOfOpenCloseRange, spread):? # Store the percent of open - close range to take as the execution price self.fractionOfOpenCloseRange = fractionOfOpenCloseRange? # Store bid/ask spread self.spread = spread? def process_order(self, data, order): # Apply fractional slippage openPrice = data.current(order.sid, 'open') closePrice = data.current(order.sid, 'close') ocRange = closePrice - openPrice ocRange = ocRange * self.fractionOfOpenCloseRange targetExecutionPrice = openPrice + ocRange log.info('\nOrder:{0} open:{1} close:{2} exec:{3} side:{4}'.format( order.sid, openPrice, closePrice, targetExecutionPrice, order.direction))? # Apply spread slippage targetExecutionPrice += self.spread * order.direction? # Create the transaction using the new price we've calculated. return (targetExecutionPrice, order.amount)

交易员Robb在2.5年内使用AUM $ 100K的条件下进行回测的Algo结果如下：

总回报率：81.49％

基准回报率：17％

Alpha：0.17

Beta：0.47

Sharpe：1.45

Sortino：2.38

波动率：0.17

最大跌幅：-13％

以上

作者：修恩

▎系列阅读

『声明：作者对提及的任何产品都没有既得利益，修恩笔记所有文章仅供参考，不构成任何投资建议策略。』

据说长得好看的人都点了????

—-

原文链接：https://news.huoxing24.com/newsdetail/20200120180814303818.html

编译者/作者：修恩笔记

玩币族申明：玩币族作为开放的资讯翻译/分享平台，所提供的所有资讯仅代表作者个人观点，与玩币族平台立场无关，且不构成任何投资理财建议。文章版权归原作者所有。