| Overall Statistics |
|
Total Orders 1835 Average Win 0.71% Average Loss -0.39% Compounding Annual Return 21.117% Drawdown 21.700% Expectancy 0.456 Start Equity 100000 End Equity 680700.43 Net Profit 580.700% Sharpe Ratio 0.805 Sortino Ratio 0.925 Probabilistic Sharpe Ratio 35.413% Loss Rate 48% Win Rate 52% Profit-Loss Ratio 1.81 Alpha 0.084 Beta 0.538 Annual Standard Deviation 0.161 Annual Variance 0.026 Information Ratio 0.287 Tracking Error 0.156 Treynor Ratio 0.241 Total Fees $4929.49 Estimated Strategy Capacity $0 Lowest Capacity Asset TSM R735QTJ8XC9X Portfolio Turnover 4.30% Drawdown Recovery 574 |
from AlgorithmImports import *
class FundamentalValueSentiment30DayRebalanceAlgorithm(QCAlgorithm):
"""
Quality-Value + NLP Sentiment + SPY 200MA Regime Filter
─────────────────────────────────────────────────────────
Universe : P/E 5-18, ROI > 12%, D/E < 1.0, div yield > 1%
Ranking : ROI sort → FinBERT/keyword sentiment rerank
Filters : momentum ROC63 > 0, sentiment floor EWMA >= 0
Risk : 15% stop loss, 50% take profit, 10-day min hold
Regime : scale to 25% when SPY < 200-day SMA
DD Guard : scale to 25% when monthly drawdown > 10%
Gold : 15% GLD when portfolio DD > 5% AND GLD ROC63 > 5%
10Y Backtest (2016-2026): CAR 21.1% | Sharpe 0.805 | DD 21.7%
Alpha 0.084 | Beta 0.538 | PSR 35.4%
Live OOS : 3M 16.2% | 1Y Sharpe 2.39 | 5Y CAGR 22.6%
Brokerage: Interactive Brokers Margin
Sentiment: FinBERT (live) / keyword fallback (backtest)
Universe refresh: monthly | Rebalance: first trading day of month
"""
# ── Sentiment ──────────────────────────────────────────────────────────
MIN_NEWS_COUNT = 3 # min articles for "trusted" sentiment classification
SENTIMENT_ALPHA = 0.2 # EWMA alpha — new article weight
DECAY_FACTOR = 1.0 - SENTIMENT_ALPHA # daily score decay (0.80)
# ── Risk ───────────────────────────────────────────────────────────────
TAKE_PROFIT = 0.50 # exit at +50% from entry
STOP_LOSS = 0.15 # exit at -15% from entry
MIN_HOLD_DAYS = 10 # grace period before stop loss is checked
# ── Drawdown Guard ─────────────────────────────────────────────────────
DD_GUARD_THRESHOLD = 0.10 # monthly DD threshold to fire circuit breaker
DD_GUARD_SCALE = 0.25 # scale all positions to this weight on fire
# ── Portfolio Construction ─────────────────────────────────────────────
SENTIMENT_ENTRY_FLOOR = 0.0 # min EWMA sentiment for trusted symbols
MAX_POSITION_WEIGHT = 0.20 # max single position weight
MAX_POSITIONS = 10 # max simultaneous equity positions
MIN_HISTORY_DAYS = 5 # min days in universe before rebalance-eligible
# ── Momentum ───────────────────────────────────────────────────────────
MOMENTUM_LOOKBACK = 63 # ROC lookback in trading days (~3 months)
MOMENTUM_MIN_RETURN = 0.0 # min ROC63 to pass momentum filter
# ── Gold ───────────────────────────────────────────────────────────────
GOLD_MAX_WEIGHT = 0.15 # max GLD allocation when conditions met
GOLD_MOMENTUM_MIN = 0.05 # min GLD ROC63 to allocate gold
# ── Queue / Cache ──────────────────────────────────────────────────────
QUEUE_MAX_SIZE = 500 # max queued news items before oldest dropped
SCORE_CACHE_MAX = 2000 # max cached sentiment scores (deduplication)
# ── Warmup ─────────────────────────────────────────────────────────────
PREWARM_MAX_ARTICLES_PER_SYMBOL = 5 # cap per symbol to prevent OOM
PREWARM_KEYWORD_ONLY = False # True = skip FinBERT in pre-warm
# ══════════════════════════════════════════════════════════════════════════
# KEYWORD SENTIMENT DICTIONARIES
# Fallback scorer when FinBERT unavailable (backtest mode).
# Tiers: STRONG=1.5 | NORMAL=1.0 | WEAK=0.5 | AMBIGUOUS=-0.3
# Negation window: 3 tokens. Macro headlines (>=3 hits) → neutral 0.0.
# Final score normalised by total weight, clamped to [-1, +1].
# ══════════════════════════════════════════════════════════════════════════
_KW_STRONG_POS: frozenset = frozenset({
"beat", "beats", "beating", "blowout", "smashed", "crushed", "topped",
"exceeded", "surpassed", "outperformed", "record", "record-breaking",
"all-time-high", "explosive", "blockbuster", "landmark", "milestone",
"raised", "raises", "raise", "lifted", "increased", "boosted", "upped",
"reiterated", "acquisition", "acquired", "merger", "buyout", "takeover",
"deal", "partnership", "collaboration", "alliance", "joint-venture",
"dividend", "dividends", "buyback", "repurchase", "special-dividend",
"distribution", "upgrade", "upgraded", "upgrades", "overweight",
"outperform", "strong-buy", "initiates", "approved", "approval",
"launched", "breakthrough", "patent", "clearance", "fda-approval",
"authorized",
})
_KW_NORMAL_POS: frozenset = frozenset({
"profit", "profits", "profitable", "earnings", "revenue", "growth",
"grew", "grow", "growing", "gains", "gain", "positive", "strong",
"solid", "robust", "healthy", "improved", "improvement", "improving",
"momentum", "expansion", "expanding", "efficient", "efficiency",
"streamlined", "optimized", "margin", "margins", "cash-flow",
"cashflow", "synergies", "synergy", "market-share", "competitive",
"dominance", "leading", "leader", "innovative", "innovation",
"differentiated", "guidance", "outlook", "forecast", "confident",
"confidence", "optimistic", "opportunity", "opportunities", "debt-free",
"investment-grade", "upgraded-credit", "liquidity", "well-capitalized",
"win", "wins", "winning", "success", "successful", "deliver",
"delivered", "delivering", "buy", "bullish", "bull",
})
_KW_WEAK_POS: frozenset = frozenset({
"stable", "steady", "maintained", "maintains", "in-line", "inline",
"met", "meets", "meeting", "matched", "matching", "resilient",
"recovery", "recovering", "stabilizing", "stabilized", "bottomed",
"rebound", "rebounding", "bouncing", "normalizing", "normalize",
"gradual", "gradually", "progress", "progressing",
})
_KW_STRONG_NEG: frozenset = frozenset({
"miss", "misses", "missed", "missed-estimates", "shortfall",
"disappointed", "disappoints", "disappointing", "dismal",
"lawsuit", "sued", "litigation", "indicted", "fraud", "scandal",
"investigation", "probe", "subpoena", "regulatory-action", "fine",
"fined", "penalty", "penalties", "violation", "violations",
"criminal", "charges", "charged", "downgrade", "downgraded",
"downgrades", "underweight", "underperform", "sell", "strong-sell",
"avoid", "bankruptcy", "bankrupt", "insolvent", "default", "defaulted",
"restructuring", "chapter-11", "liquidation", "seized", "receivership",
"collapse", "collapsed", "imploded", "cut", "cuts", "cutting",
"slashed", "slashing", "slashed-guidance", "reduced", "reduces",
"lowered", "withdrew", "withdrawn", "suspended", "suspends",
"suspending", "layoffs", "layoff", "fired", "termination",
"terminated", "mass-layoff", "job-cuts", "redundancies",
})
_KW_NORMAL_NEG: frozenset = frozenset({
"loss", "losses", "losing", "deficit", "write-down", "writedown",
"write-off", "writeoff", "impairment", "charge", "charges",
"negative", "weak", "weakness", "softness", "soft", "sluggish",
"slowdown", "slowing", "declined", "declines", "declining",
"decreased", "decrease", "fell", "fall", "falls", "falling",
"down", "drop", "drops", "dropped", "lower", "lowered",
"warned", "warns", "warning", "cautious", "caution", "headwinds",
"headwind", "pressure", "pressured", "pressures", "challenged",
"challenges", "difficult", "difficulties", "lost", "losing",
"market-share-loss", "competition", "competitive-pressure",
"disrupted", "disruption", "obsolete", "dilution", "diluted",
"debt", "leverage", "overleveraged", "downgraded-credit", "junk",
"high-yield-risk", "bear", "bearish",
})
_KW_WEAK_NEG: frozenset = frozenset({
"below", "missed-slightly", "slightly-below", "modestly-below",
"modest-decline", "slight-decline", "marginal-decline", "uncertainty",
"uncertain", "unclear", "remains-unclear", "mixed", "uneven",
"inconsistent", "volatile", "volatility", "delayed", "delay",
"delays", "postponed", "postponement", "slower", "slowed", "muted",
"subdued", "tepid", "lackluster",
})
_KW_AMBIGUOUS: frozenset = frozenset({
"volatile", "volatility", "cautious", "caution", "mixed", "uncertain",
"uncertainty", "unclear", "challenging", "complex", "complicated",
"evolving", "fluid", "dynamic", "transitioning", "transition",
"restructure", "restructuring", "transforming", "transformation",
"pivoting", "pivot",
})
_KW_NEGATIONS: frozenset = frozenset({
"not", "no", "never", "neither", "nor", "without", "lack", "lacking",
"lacks", "failed", "fails", "unable", "unlikely", "didn't", "doesn't",
"don't", "won't", "wasn't", "weren't", "isn't", "aren't", "hasn't",
"haven't", "couldn't", "wouldn't", "shouldn't", "cannot", "cant",
})
_KW_MACRO_EXCLUDE: frozenset = frozenset({
"fed", "federal-reserve", "fomc", "interest-rate", "interest-rates",
"inflation", "cpi", "ppi", "gdp", "unemployment", "jobs-report",
"nonfarm", "payrolls", "treasury", "yield-curve", "quantitative",
"tightening", "tapering", "rate-hike", "rate-cut", "basis-points",
"recession", "economic", "economy", "macro", "geopolitical",
"tariff", "tariffs", "trade-war", "sanctions", "opec",
})
# ══════════════════════════════════════════════════════════════════════════
# INITIALIZE
# ══════════════════════════════════════════════════════════════════════════
def initialize(self) -> None:
if not self.live_mode:
self.set_start_date(2016, 1, 1)
self.set_end_date(2026, 1, 1)
else:
self.set_start_date(self.end_date - timedelta(5 * 365))
self.set_benchmark("SPY")
self.set_cash(100_000)
self.set_brokerage_model(
BrokerageName.INTERACTIVE_BROKERS_BROKERAGE,
AccountType.MARGIN
)
self.universe_settings.resolution = Resolution.DAILY
self.universe_settings.data_normalization_mode = DataNormalizationMode.ADJUSTED
self.universe_settings.fill_data_before_start = True
self._spy = self.add_equity("SPY", Resolution.MINUTE).symbol
self.set_security_initializer(
lambda s: s.set_fill_model(ImmediateFillModel())
)
self.universe_settings.schedule.on(
self.date_rules.month_start(self._spy)
)
self._gld = self.add_equity("GLD", Resolution.DAILY).symbol
self._gld_momentum = self.ROC(self._gld, self.MOMENTUM_LOOKBACK, Resolution.DAILY)
self._spy_sma200 = self.SMA(self._spy, 200, Resolution.DAILY)
self._regime_filter_active = False
self._selected_symbols: List[Symbol] = []
self._coarse_count = 0
self._fine_count = 0
self._entry_price_by_symbol: dict = {}
self._position_entry_date: dict = {}
self._monthly_peak_value = 0.0
self._dd_guard_active = False
self._symbol_added_date: dict = {}
self._momentum: dict = {}
self._news_symbol_by_underlying: dict = {}
self._underlying_by_news_symbol: dict = {}
self._sentiment_ewma_by_symbol: dict = {}
self._sentiment_hit_count: dict = {}
self._sentiment_alpha = self.SENTIMENT_ALPHA
self._pending_liquidations = set()
self._news_queue: list = []
self._score_cache: dict = {}
self._use_local_finbert = self.live_mode
self._finbert = {}
self._finbert_ready = False
self._finbert_max_chars = 1500
self.set_warm_up(timedelta(days=200 + 30))
self._initialize_local_finbert()
self.add_universe(self._fundamental_selection)
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.after_market_open(self._spy, 10),
self._decay_sentiment,
)
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.every(timedelta(minutes=30)),
self._process_news_queue,
)
self.schedule.on(
self.date_rules.month_start(self._spy),
self.time_rules.after_market_open(self._spy, 30),
self._rebalance_if_due,
)
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.after_market_open(self._spy, 45),
self._daily_risk_check,
)
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.after_market_open(self._spy, 60),
self._portfolio_drawdown_guard,
)
self.schedule.on(
self.date_rules.every_day(self._spy),
self.time_rules.after_market_open(self._spy, 75),
self._regime_filter_check,
)
# ══════════════════════════════════════════════════════════════════════════
# UNIVERSE SELECTION
# ══════════════════════════════════════════════════════════════════════════
def _fundamental_selection(self, fundamental: List[Fundamental]) -> List[Symbol]:
"""
Monthly universe via Morningstar fundamentals.
Coarse: price > $5, volume > $10M → top 1000 by dollar volume.
Fine: P/E 5-18, D/E < 1.0, div yield > 1%, ROI > 12%.
Returns top 20 sorted by ROI descending.
"""
filtered = [
f for f in fundamental
if f.has_fundamental_data
and f.price is not None
and float(f.price) > 5
and f.dollar_volume > 10_000_000
]
top1000 = sorted(filtered, key=lambda f: f.dollar_volume, reverse=True)[:1000]
self._coarse_count = len(top1000)
selected = []
for f in top1000:
pe = self._get_float(f, [
"valuation_ratios.pe_ratio",
"valuation_ratios.peratio",
"valuation_ratios.price_earnings_ratio",
])
dte = self._get_float(f, [
"operation_ratios.total_debt_equity_ratio",
"operation_ratios.debt_to_equity",
"operation_ratios.debttoequity",
])
div_yield = self._get_float(f, [
"valuation_ratios.trailing_dividend_yield",
"valuation_ratios.dividend_yield",
"valuation_ratios.dividendyield",
])
roi = self._get_float(f, [
"operation_ratios.roi",
"operation_ratios.return_on_investment",
"operation_ratios.returnoninvesment",
"profitability_ratios.roi",
"profitability_ratios.return_on_investment",
"profitability_ratios.return_on_invested_capital",
"operation_ratios.roic",
"profitability_ratios.roic",
])
if not all(self._is_finite_number(v) for v in [pe, dte, div_yield, roi]):
continue
if pe < 5 or pe > 18: continue
if dte >= 1.0: continue
if div_yield <= 0.01: continue
if roi <= 0.12: continue
selected.append((f.symbol, float(roi)))
selected_sorted = sorted(selected, key=lambda x: x[1], reverse=True)
symbols = [x[0] for x in selected_sorted[:20]]
self._fine_count = len(selected_sorted)
if set(symbols) != set(self._selected_symbols):
self._selected_symbols = symbols
return symbols
# ══════════════════════════════════════════════════════════════════════════
# SECURITIES CHANGED
# ══════════════════════════════════════════════════════════════════════════
def on_securities_changed(self, changes: SecurityChanges) -> None:
"""
Removal: liquidate (or queue if warming up), clean all state.
Addition: record entry date, create ROC indicator, subscribe TiingoNews.
SPY and GLD are permanent — never removed or re-initialised here.
"""
for security in changes.removed_securities:
symbol = security.symbol
if symbol in (self._gld, self._spy):
continue
if self.is_warming_up:
self._pending_liquidations.add(symbol)
self._entry_price_by_symbol.pop(symbol, None)
self._sentiment_ewma_by_symbol.pop(symbol, None)
self._sentiment_hit_count.pop(symbol, None)
self._momentum.pop(symbol, None)
self._symbol_added_date.pop(symbol, None)
continue
if self.portfolio[symbol].invested:
self.liquidate(symbol)
self._entry_price_by_symbol.pop(symbol, None)
self._sentiment_ewma_by_symbol.pop(symbol, None)
self._sentiment_hit_count.pop(symbol, None)
self._momentum.pop(symbol, None)
self._symbol_added_date.pop(symbol, None)
self._remove_tiingo_news_subscription(symbol)
for security in changes.added_securities:
symbol = security.symbol
if symbol in (self._gld, self._spy):
continue
if symbol not in self._symbol_added_date:
self._symbol_added_date[symbol] = self.time
if symbol not in self._momentum:
self._momentum[symbol] = self.ROC(
symbol, self.MOMENTUM_LOOKBACK, Resolution.DAILY
)
self._ensure_tiingo_news_subscription(symbol)
# ══════════════════════════════════════════════════════════════════════════
# ON DATA
# ══════════════════════════════════════════════════════════════════════════
def on_data(self, slice: Slice) -> None:
"""Queue TiingoNews items for async scoring. Never scores inline."""
if slice is None:
return
try:
news_by_symbol = slice.get(TiingoNews)
except Exception:
news_by_symbol = None
if news_by_symbol is None:
return
for kvp in news_by_symbol:
try:
news_symbol = kvp.key
item = kvp.value
except Exception:
continue
if news_symbol is None or item is None:
continue
if self._underlying_by_news_symbol.get(news_symbol) is None:
continue
if len(self._news_queue) >= self.QUEUE_MAX_SIZE:
self._news_queue.pop(0)
self._news_queue.append((news_symbol, item))
# ══════════════════════════════════════════════════════════════════════════
# NEWS QUEUE PROCESSOR
# ══════════════════════════════════════════════════════════════════════════
def _process_news_queue(self) -> None:
"""
Score queued TiingoNews every 30 min via scheduler.
Priority: cache hit → FinBERT → keyword fallback.
Skipped during warmup — on_warmup_finished() handles pre-warm.
"""
if self.is_warming_up:
self._news_queue.clear()
return
if not self._news_queue:
return
snapshot = list(self._news_queue)
self._news_queue = []
scored = 0
for news_symbol, item in snapshot:
underlying = self._underlying_by_news_symbol.get(news_symbol)
if underlying is None:
continue
text = self._extract_text(item)
if not text:
continue
text_hash = hash(text)
if text_hash in self._score_cache:
score = self._score_cache[text_hash]
else:
score = self._finbert_sentiment_score(text)
if score is None:
score = self._compute_naive_text_sentiment(item)
if len(self._score_cache) >= self.SCORE_CACHE_MAX:
try:
self._score_cache.pop(next(iter(self._score_cache)))
except Exception:
pass
self._score_cache[text_hash] = score
if score is not None and self._is_finite_number(score):
self._update_sentiment(underlying, float(score))
scored += 1
if scored > 0 and self.live_mode:
self.debug(f"Queue processed: {len(snapshot)} items, {scored} scored")
# ══════════════════════════════════════════════════════════════════════════
# TEXT EXTRACTION
# ══════════════════════════════════════════════════════════════════════════
def _extract_text(self, news_item) -> str:
"""Concatenate all text fields from TiingoNews item."""
parts = []
for attr in ["title", "Title", "headline", "Headline",
"description", "Description", "summary", "Summary"]:
if hasattr(news_item, attr):
try:
val = getattr(news_item, attr)
if val:
parts.append(str(val).strip())
except Exception:
continue
return " ".join(parts).strip()
# ══════════════════════════════════════════════════════════════════════════
# ORDER EVENTS
# ══════════════════════════════════════════════════════════════════════════
def on_order_event(self, order_event: OrderEvent) -> None:
"""Record entry price/date on first fill; clear records on close."""
if order_event is None or order_event.status != OrderStatus.FILLED:
return
symbol = order_event.symbol
if symbol is None or not self.securities.contains_key(symbol):
return
holding = self.portfolio[symbol]
if holding.invested and symbol not in self._entry_price_by_symbol:
fill_price = float(order_event.fill_price)
if self._is_finite_number(fill_price) and fill_price > 0:
self._entry_price_by_symbol[symbol] = fill_price
self._position_entry_date[symbol] = self.time
if not holding.invested:
self._entry_price_by_symbol.pop(symbol, None)
self._position_entry_date.pop(symbol, None)
# ══════════════════════════════════════════════════════════════════════════
# SCHEDULED JOBS
# ══════════════════════════════════════════════════════════════════════════
def _spy_below_200ma(self) -> bool:
"""True if SPY price < 200-day SMA. False if indicator not ready."""
if not self._spy_sma200.is_ready:
return False
return float(self.securities[self._spy].price) < float(self._spy_sma200.current.value)
def _gold_target_weight(self) -> float:
"""
Returns GOLD_MAX_WEIGHT only when BOTH conditions met:
1. Portfolio drawdown from monthly peak > 5%
2. GLD ROC63 > GOLD_MOMENTUM_MIN (5%)
Returns 0.0 otherwise — avoids persistent bull-market drag.
"""
if not self._gld_momentum.is_ready:
return 0.0
if float(self._gld_momentum.current.value) <= self.GOLD_MOMENTUM_MIN:
return 0.0
equity = self.portfolio.total_portfolio_value
if self._monthly_peak_value <= 0:
return 0.0
drawdown = (self._monthly_peak_value - equity) / self._monthly_peak_value
if drawdown < 0.05:
return 0.0
return self.GOLD_MAX_WEIGHT
def _regime_filter_check(self) -> None:
"""
SPY 200MA regime filter — runs at 75 min after open.
Crossover below: scale all positions to DD_GUARD_SCALE (25%).
Suppresses DD guard (_dd_guard_active = True).
Recovery above: clear flag — positions rebuilt at next rebalance.
Fires once per crossover event, not on every bar below 200MA.
"""
if self.is_warming_up:
return
below_200ma = self._spy_below_200ma()
if below_200ma and not self._regime_filter_active:
self._regime_filter_active = True
self._dd_guard_active = True
equity = self.portfolio.total_portfolio_value
regime_msg = (
f"Regime filter triggered: SPY below 200MA | "
f"SPY={self.securities[self._spy].price:.2f} "
f"SMA200={self._spy_sma200.current.value:.2f} | "
f"scaling to {self.DD_GUARD_SCALE:.0%}"
)
self.debug(regime_msg)
if self.live_mode:
self.log(f"[REGIME] {regime_msg}")
targets = []
for symbol, holding in self.portfolio.items():
if not holding.invested:
continue
current_weight = holding.holdings_value / equity
scaled_weight = current_weight * self.DD_GUARD_SCALE
targets.append(PortfolioTarget(symbol, scaled_weight))
if targets:
self.set_holdings(targets)
elif not below_200ma and self._regime_filter_active:
self._regime_filter_active = False
recovery_msg = (
f"Regime filter cleared: SPY above 200MA | "
f"SPY={self.securities[self._spy].price:.2f} "
f"SMA200={self._spy_sma200.current.value:.2f} | "
f"positions restore at next rebalance"
)
self.debug(recovery_msg)
if self.live_mode:
self.log(f"[REGIME] {recovery_msg}")
def _rebalance_if_due(self) -> None:
"""
Monthly rebalance on first trading day of each month.
Filters: price data, min history days, momentum, sentiment floor.
Weights: sentiment-tilted 60/40, 20% position cap, gold-aware budget.
Resets monthly peak, DD guard and regime state after execution.
"""
if self.is_warming_up:
return
if not self._selected_symbols:
self.debug("Rebalance skipped — no symbols in universe yet")
return
ranked = self._rank_by_sentiment(self._selected_symbols)
if not ranked:
return
momentum_excluded = 0
history_excluded = 0
sentiment_excluded = 0
tradeable = []
for s in ranked:
if s not in self.securities:
continue
sec = self.securities[s]
if not sec.has_data or sec.price <= 0 or not sec.is_tradable:
continue
added = self._symbol_added_date.get(s)
if added is not None:
if (self.time - added).days < self.MIN_HISTORY_DAYS:
history_excluded += 1
continue
roc = self._momentum.get(s)
if roc is not None and roc.is_ready:
if float(roc.current.value) < self.MOMENTUM_MIN_RETURN:
momentum_excluded += 1
continue
hits = self._sentiment_hit_count.get(s, 0)
if hits >= self.MIN_NEWS_COUNT:
if self._get_current_sentiment(s) < self.SENTIMENT_ENTRY_FLOOR:
sentiment_excluded += 1
continue
tradeable.append(s)
if not tradeable:
self.debug("Rebalance skipped — no symbols passed filters")
return
if self.live_mode:
no_data = len(ranked) - len(tradeable) - momentum_excluded - history_excluded - sentiment_excluded
if no_data > 0: self.debug(f"Rebalance: {no_data} skipped (no price data)")
if history_excluded: self.debug(f"Rebalance: {history_excluded} skipped (< {self.MIN_HISTORY_DAYS} days)")
if momentum_excluded: self.debug(f"Rebalance: {momentum_excluded} excluded (momentum)")
if sentiment_excluded: self.debug(f"Rebalance: {sentiment_excluded} excluded (sentiment)")
ranked_f = tradeable[:self.MAX_POSITIONS]
gold_weight = self._gold_target_weight()
targets = self._build_weighted_targets(ranked_f, gold_weight)
if not targets:
return
if gold_weight > 0.0:
targets.append(PortfolioTarget(self._gld, gold_weight))
self.debug(f"Gold allocated: {gold_weight:.0%} | GLD ROC63={self._gld_momentum.current.value:.3f}")
else:
if self.portfolio[self._gld].invested:
self.liquidate(self._gld)
self.debug("Gold exited — conditions not met")
self.set_holdings(targets)
self._monthly_peak_value = self.portfolio.total_portfolio_value
self._dd_guard_active = False
self._regime_filter_active = False
self._last_rebalance_time = self.time
self._pending_rebalance = False
preview = ",".join([x.value for x in ranked_f[:5]])
scores_summary = " | ".join([
f"{s.value}={self._get_current_sentiment(s):.2f}" for s in ranked_f[:5]
])
rebalance_msg = (
f"Rebalance {self.time.date()} | "
f"coarse={self._coarse_count} fine={self._fine_count} "
f"selected={len(self._selected_symbols)} "
f"hist_excl={history_excluded} mom_excl={momentum_excluded} "
f"sent_excl={sentiment_excluded} positions={len(ranked_f)} "
f"gold={gold_weight:.0%} | top5={preview} | scores={scores_summary}"
)
self.debug(rebalance_msg)
if self.live_mode:
self.log(f"[REBALANCE] {rebalance_msg}")
for s in ranked_f:
self.log(
f" {s.value}: sentiment={self._get_current_sentiment(s):.3f} "
f"hits={self._sentiment_hit_count.get(s, 0)} "
f"finbert={'yes' if self._finbert_ready else 'keyword'}"
)
def _daily_risk_check(self) -> None:
"""
Check stop loss / take profit for each invested position.
MIN_HOLD_DAYS grace period prevents whipsaw exits after entry.
"""
if self.is_warming_up:
return
for symbol in list(self._entry_price_by_symbol.keys()):
if not self.portfolio[symbol].invested:
self._entry_price_by_symbol.pop(symbol, None)
self._position_entry_date.pop(symbol, None)
continue
entry = self._entry_price_by_symbol.get(symbol, 0.0)
if not self._is_finite_number(entry) or float(entry) <= 0:
self._entry_price_by_symbol.pop(symbol, None)
self._position_entry_date.pop(symbol, None)
continue
entry_date = self._position_entry_date.get(symbol)
if entry_date is not None:
if (self.time - entry_date).days < self.MIN_HOLD_DAYS:
continue
price = float(self.securities[symbol].price)
if not self._is_finite_number(price) or price <= 0:
continue
hit_tp = price >= (1.0 + self.TAKE_PROFIT) * float(entry)
hit_sl = price <= (1.0 - self.STOP_LOSS) * float(entry)
if hit_tp or hit_sl:
reason = "take-profit" if hit_tp else "stop-loss"
exit_msg = f"Risk exit [{reason}] {symbol} | entry={entry:.2f} now={price:.2f}"
self.debug(exit_msg)
if self.live_mode:
self.log(f"[RISK EXIT] {exit_msg}")
self.liquidate(symbol)
self._entry_price_by_symbol.pop(symbol, None)
self._position_entry_date.pop(symbol, None)
def _portfolio_drawdown_guard(self) -> None:
"""
Circuit breaker: scale to DD_GUARD_SCALE if monthly DD > DD_GUARD_THRESHOLD.
Skipped if regime filter already active (regime takes priority).
Fires once per month — resets at next rebalance.
"""
if self.is_warming_up:
return
if self._regime_filter_active:
return
equity = self.portfolio.total_portfolio_value
if self._monthly_peak_value <= 0:
self._monthly_peak_value = equity
return
if equity > self._monthly_peak_value:
self._monthly_peak_value = equity
return
if self._dd_guard_active:
return
drawdown = (self._monthly_peak_value - equity) / self._monthly_peak_value
if drawdown < self.DD_GUARD_THRESHOLD:
return
self._dd_guard_active = True
guard_msg = (
f"DD Guard triggered: drawdown={drawdown:.1%} from peak "
f"${self._monthly_peak_value:,.0f} | current=${equity:,.0f} | "
f"scaling to {self.DD_GUARD_SCALE:.0%}"
)
self.debug(guard_msg)
if self.live_mode:
self.log(f"[DD GUARD] {guard_msg}")
targets = []
for symbol, holding in self.portfolio.items():
if not holding.invested:
continue
current_weight = holding.holdings_value / equity
targets.append(PortfolioTarget(symbol, current_weight * self.DD_GUARD_SCALE))
if targets:
self.set_holdings(targets)
def _decay_sentiment(self) -> None:
"""Daily EWMA decay: score *= DECAY_FACTOR. Prevents stale sentiment dominance."""
if self.is_warming_up:
return
for symbol in list(self._sentiment_ewma_by_symbol.keys()):
v = self._sentiment_ewma_by_symbol.get(symbol)
if self._is_finite_number(v):
self._sentiment_ewma_by_symbol[symbol] = float(v) * self.DECAY_FACTOR
# ══════════════════════════════════════════════════════════════════════════
# SENTIMENT RANKING & WEIGHTING
# ══════════════════════════════════════════════════════════════════════════
def _rank_by_sentiment(self, symbols: List[Symbol]) -> List[Symbol]:
"""
Rank by EWMA sentiment score.
Trusted symbols (>= MIN_NEWS_COUNT articles) ranked first,
untrusted symbols follow — both sorted descending by score.
"""
if not symbols:
return []
trusted, untrusted = [], []
for s in symbols:
score = float(self._get_current_sentiment(s))
hits = self._sentiment_hit_count.get(s, 0)
(trusted if hits >= self.MIN_NEWS_COUNT else untrusted).append((s, score))
trusted.sort(key=lambda x: x[1], reverse=True)
untrusted.sort(key=lambda x: x[1], reverse=True)
return [x[0] for x in trusted + untrusted]
def _build_weighted_targets(self, ranked: List[Symbol], gold_weight: float = 0.0) -> List[PortfolioTarget]:
"""
Sentiment-tilted 60/40 weighting within equity_budget = 1 - gold_weight.
Top 25% by sentiment → 60% of budget. Bottom 75% → 40%.
Per-position cap: MAX_POSITION_WEIGHT. Renormalise after capping.
"""
n = len(ranked)
if n <= 0:
return []
equity_budget = 1.0 - gold_weight
top_n = max(1, min(int(math.ceil(0.25 * n)), n))
rest_n = n - top_n
weights: dict = {}
if rest_n <= 0:
w = equity_budget / n
for s in ranked:
weights[s] = w
else:
w_top = (0.60 * equity_budget) / top_n
w_rest = (0.40 * equity_budget) / rest_n
for i, s in enumerate(ranked):
weights[s] = w_top if i < top_n else w_rest
for s in weights:
if weights[s] > self.MAX_POSITION_WEIGHT:
weights[s] = self.MAX_POSITION_WEIGHT
total = sum(weights.values())
if self._is_finite_number(total) and total > 0 and abs(total - equity_budget) > 1e-6:
scale = equity_budget / total
for s in weights:
weights[s] *= scale
return [PortfolioTarget(s, float(w)) for s, w in weights.items()]
# ══════════════════════════════════════════════════════════════════════════
# SENTIMENT HELPERS
# ══════════════════════════════════════════════════════════════════════════
def _update_sentiment(self, symbol: Symbol, score: float) -> None:
"""EWMA update: new = alpha*score + (1-alpha)*prev. Increments hit count."""
if symbol is None or not self._is_finite_number(score):
return
prev = self._sentiment_ewma_by_symbol.get(symbol, float("nan"))
if not self._is_finite_number(prev):
self._sentiment_ewma_by_symbol[symbol] = float(score)
else:
a = self._sentiment_alpha
self._sentiment_ewma_by_symbol[symbol] = (
a * float(score) + (1.0 - a) * float(prev)
)
self._sentiment_hit_count[symbol] = self._sentiment_hit_count.get(symbol, 0) + 1
def _get_current_sentiment(self, symbol: Symbol) -> float:
"""Returns current EWMA sentiment score. Defaults to 0.0 if no data."""
if symbol is None:
return 0.0
v = self._sentiment_ewma_by_symbol.get(symbol, float("nan"))
return float(v) if self._is_finite_number(v) else 0.0
# ══════════════════════════════════════════════════════════════════════════
# FINBERT
# ══════════════════════════════════════════════════════════════════════════
def _initialize_local_finbert(self) -> None:
"""
Load ProsusAI/finbert pipeline for live mode only.
Skipped in backtest — 200ms/article × 100k+ articles = 50+ hour runtime.
Runs validation inference on load; sets _finbert_ready = True on success.
"""
self._finbert_ready = False
self._finbert = {}
if not self.live_mode:
self.debug("FinBERT skipped — backtest mode, using keyword model")
return
if not self._use_local_finbert:
self.debug("FinBERT disabled by configuration")
return
try:
from transformers import pipeline # type: ignore
pipe = pipeline(
task="sentiment-analysis",
model="ProsusAI/finbert",
tokenizer="ProsusAI/finbert",
truncation=True,
)
test = pipe("earnings beat expectations")
if not test:
raise RuntimeError("Empty test inference")
self._finbert = {"pipeline": pipe}
self._finbert_ready = True
self.debug(f"Local FinBERT ready (test: {test[0].get('label')})")
except Exception as exc:
self.debug(f"Local FinBERT unavailable: {exc}")
def _finbert_sentiment_score(self, text: str):
"""
Run FinBERT inference. Returns float in [-1, +1] or None on failure.
positive → +confidence | negative → -confidence | neutral → 0.0
"""
if not self._finbert_ready or not text:
return None
try:
pipe = self._finbert.get("pipeline")
if pipe is None:
return None
result = pipe(str(text).strip()[:self._finbert_max_chars])
if isinstance(result, list) and result:
result = result[0]
if not isinstance(result, dict):
return None
label = str(result.get("label", "")).lower()
conf = max(0.0, min(1.0, float(result.get("score", 0.0))))
if "pos" in label: return conf
if "neg" in label: return -conf
if "neu" in label: return 0.0
return None
except Exception:
return None
# ══════════════════════════════════════════════════════════════════════════
# KEYWORD SENTIMENT FALLBACK
# ══════════════════════════════════════════════════════════════════════════
def _compute_naive_text_sentiment(self, news_item) -> float:
"""
Weighted keyword scorer — three tiers (1.5/1.0/0.5), negation window
of 3 tokens, ambiguous words = -0.3, macro exclusion filter.
Score = total_score / total_weight, clamped to [-1, +1].
Returns None if text too short (< 3 tokens) or extraction fails.
"""
text = self._extract_text(news_item)
if not text:
return None
try:
cleaned = str(text).lower()
for ch in "\n\r\t,.;:!?()[]{}'\"":
cleaned = cleaned.replace(ch, " ")
tokens = [t.strip("-") for t in cleaned.split() if t.strip("-")]
except Exception:
return None
if len(tokens) < 3:
return None
bigrams = [tokens[i] + "-" + tokens[i+1] for i in range(len(tokens)-1)]
trigrams = [tokens[i] + "-" + tokens[i+1] + "-" + tokens[i+2]
for i in range(len(tokens)-2)]
all_tokens = tokens + bigrams + trigrams
if sum(1 for t in all_tokens if t in self._KW_MACRO_EXCLUDE) >= 3:
return 0.0
total_score = 0.0
total_weight = 0.0
negation_indices = {i for i, t in enumerate(tokens) if t in self._KW_NEGATIONS}
def _is_negated(idx: int) -> bool:
return any(idx - 3 <= ni < idx for ni in negation_indices)
for i, token in enumerate(tokens):
weight = polarity = None
if token in self._KW_STRONG_POS: weight, polarity = 1.5, 1.0
elif token in self._KW_NORMAL_POS: weight, polarity = 1.0, 1.0
elif token in self._KW_WEAK_POS: weight, polarity = 0.5, 1.0
elif token in self._KW_STRONG_NEG: weight, polarity = 1.5, -1.0
elif token in self._KW_NORMAL_NEG: weight, polarity = 1.0, -1.0
elif token in self._KW_WEAK_NEG: weight, polarity = 0.5, -1.0
elif token in self._KW_AMBIGUOUS:
total_score -= 0.3
total_weight += 0.3
continue
if weight is None:
continue
if _is_negated(i):
polarity *= -1.0
total_score += weight * polarity
total_weight += weight
if total_weight == 0.0:
return 0.0
score = max(-1.0, min(1.0, total_score / total_weight))
return float(score) if self._is_finite_number(score) else None
# ══════════════════════════════════════════════════════════════════════════
# TIINGO NEWS SUBSCRIPTIONS
# ══════════════════════════════════════════════════════════════════════════
def _ensure_tiingo_news_subscription(self, underlying: Symbol) -> None:
"""
Subscribe to TiingoNews for an equity symbol.
is_tradable=False prevents news tickers appearing in capacity calculation
(known QC bug: news symbols with zero liquidity cause capacity = $0).
"""
if underlying is None or underlying in self._news_symbol_by_underlying:
return
news_security = self.add_data(TiingoNews, underlying)
news_security.is_tradable = False
news_symbol = news_security.symbol
self._news_symbol_by_underlying[underlying] = news_symbol
self._underlying_by_news_symbol[news_symbol] = underlying
def _remove_tiingo_news_subscription(self, underlying: Symbol) -> None:
"""Remove TiingoNews subscription and clean up both lookup maps."""
if underlying is None:
return
news_symbol = self._news_symbol_by_underlying.pop(underlying, None)
if news_symbol is None:
return
self._underlying_by_news_symbol.pop(news_symbol, None)
try:
self.remove_security(news_symbol)
except Exception:
pass
# ══════════════════════════════════════════════════════════════════════════
# STARTUP RECONCILIATION
# ══════════════════════════════════════════════════════════════════════════
def _reconcile_existing_holdings(self) -> None:
"""
Adopt existing account positions on strategy restart.
Populates entry price from IB average cost and sets entry date to
MIN_HOLD_DAYS ago so stop loss activates immediately.
Skips GLD, SPY, and symbols already tracked.
Called after pending liquidations are cleared in on_warmup_finished().
"""
reconciled = 0
for symbol, holding in self.portfolio.items():
if not holding.invested:
continue
if symbol in (self._gld, self._spy):
continue
if symbol in self._entry_price_by_symbol:
continue
avg_cost = float(holding.average_price)
if not self._is_finite_number(avg_cost) or avg_cost <= 0:
continue
self._entry_price_by_symbol[symbol] = avg_cost
self._position_entry_date[symbol] = (
self.time - timedelta(days=self.MIN_HOLD_DAYS)
)
reconciled += 1
self.debug(f"Reconciled existing holding: {symbol.value} avg_cost={avg_cost:.2f}")
if reconciled > 0:
msg = f"Startup reconciliation: {reconciled} existing positions adopted"
self.debug(msg)
if self.live_mode:
self.log(f"[RECONCILE] {msg}")
else:
self.debug("Startup reconciliation: no existing positions found")
# ══════════════════════════════════════════════════════════════════════════
# WARMUP FINISHED
# ══════════════════════════════════════════════════════════════════════════
def on_warmup_finished(self) -> None:
"""
Runs once when 230-day warmup ends.
1. Preserve pre-existing IB holdings from pending liquidation queue.
2. Liquidate remaining queued symbols (genuine universe dropouts).
3. Reconcile preserved holdings into entry price/date tracking.
4. Pre-warm sentiment from last 10 days of TiingoNews history.
"""
# ── 1. Preserve pre-existing holdings from liquidation queue ───────
# Symbols that dropped out of universe during warmup should not be
# liquidated if they were held in IB before restart — reconciliation
# will adopt them. Only remove them from the queue here; liquidation
# of genuine dropouts happens in step 2.
pre_existing = {
s for s in self._pending_liquidations
if self.portfolio[s].invested
and self._is_finite_number(float(self.portfolio[s].average_price))
and float(self.portfolio[s].average_price) > 0
}
if pre_existing:
names = ", ".join(s.value for s in pre_existing)
self.debug(f"Preserving pre-existing holdings from liquidation queue: {names}")
if self.live_mode:
self.log(f"[RECONCILE] Preserving pre-existing holdings: {names}")
self._pending_liquidations -= pre_existing
# ── 2. Liquidate genuine universe dropouts ─────────────────────────
for symbol in list(self._pending_liquidations):
if self.portfolio[symbol].invested:
self.liquidate(symbol)
self._entry_price_by_symbol.pop(symbol, None)
self._sentiment_ewma_by_symbol.pop(symbol, None)
self._sentiment_hit_count.pop(symbol, None)
self._momentum.pop(symbol, None)
self._symbol_added_date.pop(symbol, None)
self._remove_tiingo_news_subscription(symbol)
self._pending_liquidations.clear()
# ── 3. Reconcile existing holdings into risk tracking ──────────────
self._reconcile_existing_holdings()
# ── 4. Sentiment pre-warm ──────────────────────────────────────────
scorer = "keyword-only" if (self.PREWARM_KEYWORD_ONLY or not self.live_mode) \
else "FinBERT+keyword"
self.debug(f"Sentiment pre-warm starting [{scorer}] ...")
total_articles = 0
total_scored = 0
for underlying, news_symbol in list(self._news_symbol_by_underlying.items()):
try:
history = self.history(TiingoNews, news_symbol, 10, Resolution.DAILY)
if history is None or history.empty:
continue
rows = list(history.iterrows())[-self.PREWARM_MAX_ARTICLES_PER_SYMBOL:]
for _, row in rows:
text_parts = []
for col in ["title", "description", "summary",
"Title", "Description", "Summary"]:
val = row.get(col, "")
if val and str(val).strip():
text_parts.append(str(val).strip())
text = " ".join(text_parts).strip()
if not text:
continue
total_articles += 1
text_hash = hash(text)
if text_hash in self._score_cache:
score = self._score_cache[text_hash]
else:
score = None
if not self.PREWARM_KEYWORD_ONLY and self.live_mode:
score = self._finbert_sentiment_score(text)
if score is None:
class _Row:
pass
r = _Row()
r.title = row.get("title", "")
r.description = row.get("description", "")
r.summary = row.get("summary", "")
score = self._compute_naive_text_sentiment(r)
if len(self._score_cache) >= self.SCORE_CACHE_MAX:
try:
self._score_cache.pop(next(iter(self._score_cache)))
except Exception:
pass
self._score_cache[text_hash] = score
if score is not None and self._is_finite_number(score):
self._update_sentiment(underlying, float(score))
total_scored += 1
except Exception as e:
self.debug(f"Pre-warm error [{underlying.value}]: {e}")
continue
trusted = sum(1 for c in self._sentiment_hit_count.values() if c >= self.MIN_NEWS_COUNT)
self.debug(
f"Sentiment pre-warm complete | articles={total_articles} "
f"scored={total_scored} | symbols={len(self._sentiment_hit_count)} "
f"trusted (>={self.MIN_NEWS_COUNT} hits)={trusted}"
)
# ══════════════════════════════════════════════════════════════════════════
# STATIC UTILITIES
# ══════════════════════════════════════════════════════════════════════════
@staticmethod
def _is_finite_number(x) -> bool:
"""True if x is a finite real number — not None, NaN, Inf, or bool."""
if x is None or isinstance(x, bool):
return False
try:
return math.isfinite(float(x))
except Exception:
return False
@staticmethod
def _get_float(obj, attr_paths: List[str]):
"""
Traverse nested object via dot-notation paths, return first valid float.
Tries multiple paths — handles Morningstar field name inconsistencies.
Unwraps QC IndicatorDataPoint .value attribute if present.
"""
for path in attr_paths:
current = obj
ok = True
for part in path.split("."):
if current is None or not hasattr(current, part):
ok = False
break
current = getattr(current, part)
if not ok or current is None:
continue
if hasattr(current, "value"):
current = current.value
try:
return float(current)
except Exception:
continue
return None
def _to_ratio(self, value):
"""Convert percentage value > 1.0 to decimal ratio. Returns None if invalid."""
if not self._is_finite_number(value):
return None
v = float(value)
return v / 100.0 if v > 1.0 else v