#region imports
from AlgorithmImports import *
#endregion
class NewsSentimentAlphaModel(AlphaModel):
_securities = []
# Assign sentiment values to words
_word_scores = {'good': 1, 'great': 1, 'best': 1, 'growth': 1,
'bad': -1, 'terrible': -1, 'worst': -1, 'loss': -1}
def update(self, algorithm: QCAlgorithm, data: Slice) -> List[Insight]:
insights = []
for security in self._securities:
if not security.exchange.hours.is_open(algorithm.time + timedelta(minutes=1), extended_market_hours=False):
continue
if not data.contains_key(security.dataset_symbol):
continue
article = data[security.dataset_symbol]
# Assign a sentiment score to the article
title_words = article.description.lower()
score = 0
for word, word_score in self._word_scores.items():
if word in title_words:
score += word_score
# Only trade when there is positive news
if score > 0:
direction = InsightDirection.UP
elif score < 0:
direction = InsightDirection.FLAT
else:
continue
# Create insights
expiry = security.exchange.hours.get_next_market_close(algorithm.time, extended_market_hours=False) - timedelta(minutes=1, seconds=1)
insights.append(Insight.price(security.symbol, expiry, direction, None, None, None, 1/len(self._securities)))
return insights
def on_securities_changed(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
for security in changes.added_securities:
# Subscribe to the Tiingo News Feed for this security
security.dataset_symbol = algorithm.add_data(TiingoNews, security.symbol).symbol
self._securities.append(security)
for security in changes.removed_securities:
if security.symbol in self._securities:
# Unsubscribe from the Tiingo News Feed for this security
algorithm.remove_security(self.dataset_symbol)
self._securities.remove(security)
# region imports
from AlgorithmImports import *
from universe import FaangUniverseSelectionModel
from alpha import NewsSentimentAlphaModel
from portfolio import PartitionedPortfolioConstructionModel
# endregion
class BreakingNewsEventsAlgorithm(QCAlgorithm):
_undesired_symbols_from_previous_deployment = []
_checked_symbols_from_previous_deployment = False
def initialize(self):
self.set_end_date(datetime.now())
self.set_start_date(self.end_date - timedelta(5*365))
self.set_cash(1_000_000)
self.settings.minimum_order_margin_portfolio_percentage = 0
self.set_brokerage_model(BrokerageName.INTERACTIVE_BROKERS_BROKERAGE, AccountType.MARGIN)
self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW
universe = FaangUniverseSelectionModel()
self.add_universe_selection(universe)
self.add_alpha(NewsSentimentAlphaModel())
# We use 5 partitions because the FAANG universe has 5 members.
# If we change the universe to have, say, 100 securities, then 100 paritions means
# that each trade gets a 1% (1/100) allocation instead of a 20% (1/5) allocation.
self.set_portfolio_construction(PartitionedPortfolioConstructionModel(self, universe.count))
self.add_risk_management(NullRiskManagementModel())
self.set_execution(ImmediateExecutionModel())
def on_data(self, data):
# Exit positions that aren't backed by existing insights.
# If you don't want this behavior, delete this method definition.
if not self.is_warming_up and not self._checked_symbols_from_previous_deployment:
for security_holding in self.portfolio.values():
if not security_holding.invested:
continue
symbol = security_holding.symbol
if not self.insights.has_active_insights(symbol, self.utc_time):
self._undesired_symbols_from_previous_deployment.append(symbol)
self._checked_symbols_from_previous_deployment = True
for symbol in self._undesired_symbols_from_previous_deployment:
if self.is_market_open(symbol):
self.liquidate(symbol, tag="Holding from previous deployment that's no longer desired")
self._undesired_symbols_from_previous_deployment.remove(symbol)
#region imports
from AlgorithmImports import *
#endregion
class PartitionedPortfolioConstructionModel(PortfolioConstructionModel):
def __init__(self, algorithm, num_partitions):
self._algorithm = algorithm
self._NUM_PARTITIONS = num_partitions
# REQUIRED: Will determine the target percent for each insight
def determine_target_percent(self, activeInsights: List[Insight]) -> Dict[Insight, float]:
target_pct_by_insight = {}
# Sort insights by time they were emmited
insights_sorted_by_time = sorted(activeInsights, key=lambda x: x.generated_time_utc)
# Find target securities and group insights by Symbol
target_symbols = []
insight_by_symbol = {}
for insight in insights_sorted_by_time:
insight_by_symbol[insight.symbol] = insight
if len(target_symbols) < self._NUM_PARTITIONS:
target_symbols.append(insight.symbol)
occupied_portfolio_value = 0
occupied_partitions = 0
# Get last insight emmited for each target Symbol
for symbol, insight in insight_by_symbol.items():
# Only invest in Symbols in `target_symbols`
if symbol not in target_symbols:
target_pct_by_insight[insight] = 0
else:
security_holding = self._algorithm.portfolio[symbol]
# If we're invested in the security in the proper direction, do nothing
if (security_holding.is_short and insight.direction == InsightDirection.DOWN or
security_holding.is_long and insight.direction == InsightDirection.UP):
occupied_portfolio_value += security_holding.absolute_holdings_value
occupied_partitions += 1
continue
# If currently invested and there but the insight direction has changed,
# change portfolio weight of security and reset set partition size
if (security_holding.is_short and insight.direction == InsightDirection.UP or
security_holding.is_long and insight.direction == InsightDirection.DOWN):
target_pct_by_insight[insight] = int(insight.direction)
# If not currently invested, set portfolio weight of security with partition size
if not security_holding.invested:
target_pct_by_insight[insight] = int(insight.direction)
# Scale down target percentages to respect partitions (account for liquidations from insight expiry + universe removals)
total_portfolio_value = self._algorithm.portfolio.total_portfolio_value
free_portfolio_pct = (total_portfolio_value - occupied_portfolio_value) / total_portfolio_value
vacant_partitions = self._NUM_PARTITIONS - occupied_partitions
scaling_factor = free_portfolio_pct / vacant_partitions if vacant_partitions != 0 else 0
for insight, target_pct in target_pct_by_insight.items():
target_pct_by_insight[insight] = target_pct * scaling_factor
return target_pct_by_insight
# Determines if the portfolio should be rebalanced base on the provided rebalancing func
def is_rebalance_due(self, insights: List[Insight], algorithmUtc: datetime) -> bool:
# Rebalance when any of the following cases are true:
# Case 1: A security we're invested in was removed from the universe
# Case 2: The latest insight for a Symbol we're invested in has expired
# Case 3: The insight direction for a security we're invested in has changed
# Case 4: There is an insight for a security we're not currently invested in AND there is an available parition in the portfolio
last_active_insights = self.get_target_insights() # Warning: This assumes that all insights have the same duration
insight_symbols = [insight.symbol for insight in last_active_insights]
num_investments = 0
for symbol, security_holding in self._algorithm.portfolio.items():
if not security_holding.invested:
continue
num_investments += 1
# Case 1: A security we're invested in was removed from the universe
# Case 2: The latest insight for a Symbol we're invested in has expired
if symbol not in insight_symbols:
return True
for insight in last_active_insights:
security_holding = self._algorithm.portfolio[insight.symbol]
# Case 3: The insight direction for a security we're invested in has changed
if (security_holding.is_short and insight.direction == InsightDirection.UP or
security_holding.is_long and insight.direction == InsightDirection.DOWN):
return True
# Case 4: There is an insight for a security we're not currently invested in AND there is an available parition in the portfolio
if not security_holding.invested and num_investments < self._NUM_PARTITIONS:
return True
return False
#region imports
from AlgorithmImports import *
#endregion
# 07/13/2023: -Replaced the SymbolData class by with custom Security properties
# -Fixed warm-up logic to liquidate undesired portfolio holdings on re-deployment
# -Set the MinimumOrderMarginPortfolioPercentage to 0
# https://www.quantconnect.com/terminal/processCache?request=embedded_backtest_99f55f8195cad4c3801449cf2b5699e6.html
#
# 04/15/2024: -Updated to PEP8 style
# https://www.quantconnect.com/terminal/processCache?request=embedded_backtest_b459552222b3a180c519d650ca1597bc.html
#region imports
from AlgorithmImports import *
#endregion
class FaangUniverseSelectionModel(ManualUniverseSelectionModel):
def __init__(self):
tickers = ["META", "AAPL", "AMZN", "NFLX", "GOOGL"]
self.count = len(tickers)
symbols = [Symbol.create(ticker, SecurityType.EQUITY, Market.USA) for ticker in tickers]
super().__init__(symbols)
