| 1 | """Tests for stocktwits.py - StockTwits source (ticker/crypto topics only). |
| 2 | |
| 3 | All hermetic: symbol-search and stream HTTP calls are patched, so no network. |
| 4 | """ |
| 5 | |
| 6 | from __future__ import annotations |
| 7 | |
| 8 | from unittest.mock import patch |
| 9 | |
| 10 | from lib import stocktwits, normalize, planner, pipeline |
| 11 | |
| 12 | |
| 13 | # === Gating: is_financial_topic / detect_symbols === |
| 14 | |
| 15 | def test_cashtag_is_financial(): |
| 16 | assert stocktwits.is_financial_topic("$NVDA earnings") |
| 17 | assert stocktwits.detect_symbols("$NVDA earnings", resolve=False) == ["NVDA"] |
| 18 | |
| 19 | |
| 20 | def test_crypto_alias_resolves_with_dot_x_suffix(): |
| 21 | assert stocktwits.detect_symbols("bitcoin price", resolve=False) == ["BTC.X"] |
| 22 | assert stocktwits.detect_symbols("should I buy ethereum", resolve=False) == ["ETH.X"] |
| 23 | |
| 24 | |
| 25 | def test_non_financial_topic_resolves_to_nothing(): |
| 26 | # The whole point of the gate: a person/recipe must never resolve a ticker. |
| 27 | assert not stocktwits.is_financial_topic("Kanye West") |
| 28 | assert stocktwits.detect_symbols("Kanye West", resolve=True) == [] |
| 29 | assert stocktwits.detect_symbols("Apple pie recipe", resolve=True) == [] |
| 30 | |
| 31 | |
| 32 | def test_general_topics_with_ambiguous_words_do_not_trip_the_gate(): |
| 33 | # Regression guard for the tightened _FINANCE_HINTS: these everyday phrases |
| 34 | # contain words that USED to trip the gate (share/token/coin/bear) and must |
| 35 | # never register stocktwits for a non-financial run. |
| 36 | for topic in ( |
| 37 | "how to share files on iPhone", |
| 38 | "Claude token limits", |
| 39 | "coin collecting for beginners", |
| 40 | "bear attacks in Yellowstone", |
| 41 | "bull riding championship", |
| 42 | ): |
| 43 | assert not stocktwits.is_financial_topic(topic), topic |
| 44 | assert stocktwits.detect_symbols(topic, resolve=False) == [], topic |
| 45 | |
| 46 | |
| 47 | def test_finance_vocabulary_still_trips_the_gate(): |
| 48 | for topic in ( |
| 49 | "bullish on NVDA earnings", |
| 50 | "TSLA stock forecast", |
| 51 | "best dividend stocks", |
| 52 | "altcoin season predictions", |
| 53 | "bitcoin price", |
| 54 | ): |
| 55 | assert stocktwits.is_financial_topic(topic), topic |
| 56 | |
| 57 | |
| 58 | def test_name_resolution_only_fires_for_financial_topics(): |
| 59 | # "Apple pie recipe" trips no finance hint -> no symbol-search call at all. |
| 60 | with patch.object(stocktwits, "_get_json") as mock_get: |
| 61 | assert stocktwits.detect_symbols("Apple pie recipe", resolve=True) == [] |
| 62 | mock_get.assert_not_called() |
| 63 | |
| 64 | # "ServiceNow stock" trips the gate -> symbol search runs and resolves. |
| 65 | with patch.object(stocktwits, "_get_json", return_value={"results": [{"symbol": "NOW"}]}) as mock_get: |
| 66 | assert stocktwits.detect_symbols("ServiceNow stock", resolve=True) == ["NOW"] |
| 67 | mock_get.assert_called_once() |
| 68 | |
| 69 | |
| 70 | # === Parsing + sentiment aggregation === |
| 71 | |
| 72 | def _msg(mid, user, body, sentiment=None, likes=0, created="2026-06-20T12:00:00Z"): |
| 73 | m = { |
| 74 | "id": mid, |
| 75 | "body": body, |
| 76 | "created_at": created, |
| 77 | "user": {"username": user, "followers": 1000}, |
| 78 | "likes": {"total": likes}, |
| 79 | } |
| 80 | if sentiment: |
| 81 | m["entities"] = {"sentiment": {"basic": sentiment}} |
| 82 | return m |
| 83 | |
| 84 | |
| 85 | def _response(): |
| 86 | return { |
| 87 | "symbols": ["NOW"], |
| 88 | "watchlist": 41000, |
| 89 | "messages": [ |
| 90 | _msg(1, "alice", "$NOW buy the dip", "Bullish", likes=5), |
| 91 | _msg(2, "bob", "$NOW going to zero", "Bearish", likes=2), |
| 92 | _msg(3, "carol", "$NOW holding", None, likes=0), |
| 93 | ], |
| 94 | } |
| 95 | |
| 96 | |
| 97 | def test_aggregate_sentiment_counts_and_ratio(): |
| 98 | agg = stocktwits.aggregate_sentiment(_response()["messages"]) |
| 99 | assert agg["bullish"] == 1 |
| 100 | assert agg["bearish"] == 1 |
| 101 | assert agg["untagged"] == 1 |
| 102 | assert agg["pct_bullish"] == 50 |
| 103 | assert agg["sample"] == 3 |
| 104 | |
| 105 | |
| 106 | def test_aggregate_sentiment_no_tagged_messages(): |
| 107 | agg = stocktwits.aggregate_sentiment([_msg(1, "x", "$NOW", None)]) |
| 108 | assert agg["pct_bullish"] is None # no division by zero |
| 109 | |
| 110 | |
| 111 | def test_parse_builds_well_formed_items(): |
| 112 | items = stocktwits.parse_stocktwits_response(_response(), query="ServiceNow") |
| 113 | assert len(items) == 3 |
| 114 | first = items[0] |
| 115 | assert first["url"] == "https://stocktwits.com/alice/message/1" |
| 116 | assert first["author"] == "alice" |
| 117 | assert first["metadata"]["sentiment"] == "Bullish" |
| 118 | assert first["metadata"]["symbol"] == "NOW" |
| 119 | # The bull/bear aggregate rides on every item so synthesis can cite the ratio. |
| 120 | assert first["metadata"]["sentiment_aggregate"]["pct_bullish"] == 50 |
| 121 | assert first["engagement"]["likes"] == 5 |
| 122 | |
| 123 | |
| 124 | # === Normalize wiring === |
| 125 | |
| 126 | def test_normalizer_registered_and_maps_fields(): |
| 127 | items = stocktwits.parse_stocktwits_response(_response(), query="ServiceNow") |
| 128 | normalized = normalize.normalize_source_items( |
| 129 | "stocktwits", items, from_date="2026-06-01", to_date="2026-06-30") |
| 130 | assert len(normalized) == 3 |
| 131 | item = normalized[0] |
| 132 | assert item.source == "stocktwits" |
| 133 | assert item.container == "NOW" # symbol -> container |
| 134 | assert item.author == "alice" |
| 135 | assert item.metadata["sentiment"] == "Bullish" |
| 136 | |
| 137 | |
| 138 | # === Planner + pipeline gate === |
| 139 | |
| 140 | def test_planner_capability_and_priority(): |
| 141 | assert planner.SOURCE_CAPABILITIES["stocktwits"] == {"social", "market", "finance_social"} |
| 142 | assert "stocktwits" in planner.SOURCE_PRIORITY["breaking_news"] |
| 143 | assert "stocktwits" in planner.SOURCE_PRIORITY["prediction"] |
| 144 | |
| 145 | |
| 146 | def test_pipeline_availability_is_gated_by_financial_flag(): |
| 147 | assert "stocktwits" in pipeline.available_sources({"_financial_topic": True}) |
| 148 | assert "stocktwits" not in pipeline.available_sources({"_financial_topic": False}) |
| 149 | assert "stocktwits" not in pipeline.available_sources({}) # default off |
| 150 |