| 1 | """DripStack source for last30days — premium financial newsletter search. |
| 2 | |
| 3 | DripStack indexes paid Substack newsletters, analyst writeups, and financial |
| 4 | podcasts. The search endpoint is free and public (no API key); it returns |
| 5 | article metadata including title, publication, date, and a relevance-scored |
| 6 | snippet. Full article summaries and stock picks are behind a paid layer and |
| 7 | are out of scope for this source adapter. |
| 8 | |
| 9 | The signal is complementary to the other financial sources: StockTwits gives |
| 10 | retail sentiment, Polymarket gives prediction-market odds, and DripStack gives |
| 11 | what professional analysts and paid newsletter authors are actually writing |
| 12 | about. The search results carry publication attribution (e.g. "SemiAnalysis", |
| 13 | "Bloomberg") which is high-credibility signal for synthesis. |
| 14 | |
| 15 | GATING: DripStack search is most valuable for finance, markets, company |
| 16 | analysis, and industry research topics. Like arXiv (science) and Techmeme |
| 17 | (tech news), DripStack is relevance-gated — the search API itself filters |
| 18 | for topic match, so off-topic runs return thin results naturally and the |
| 19 | engine's thin-retry + relevance scoring handles the rest. |
| 20 | |
| 21 | API: public, no auth. Search endpoint returns up to 30 items per query. |
| 22 | """ |
| 23 | |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import datetime |
| 27 | import json |
| 28 | import re |
| 29 | import sys |
| 30 | import urllib.parse |
| 31 | from typing import Any |
| 32 | |
| 33 | from . import http |
| 34 | |
| 35 | _BASE_URL = "https://dripstack.xyz" |
| 36 | _SEARCH_URL = f"{_BASE_URL}/api/v1/search" |
| 37 | _UA = "Mozilla/5.0 (last30days dripstack source)" |
| 38 | |
| 39 | # Depth controls how many results we request per subquery. |
| 40 | _DEPTH_LIMITS = {"quick": 5, "default": 10, "deep": 20} |
| 41 | |
| 42 | |
| 43 | def _log(msg: str) -> None: |
| 44 | try: |
| 45 | from . import log as _enginelog |
| 46 | _enginelog.source_log("DripStack", msg, tty_only=False) |
| 47 | except Exception: |
| 48 | print(f"[DripStack] {msg}", file=sys.stderr) |
| 49 | |
| 50 | |
| 51 | def _get_json(url: str, timeout: int = 20) -> dict[str, Any]: |
| 52 | # All engine traffic goes through the shared lib/http.py choke point so |
| 53 | # capture/replay, fixtures, and failure taxonomy apply to this source too. |
| 54 | return http.get(url, headers={"User-Agent": _UA}, timeout=timeout, retries=2) |
| 55 | |
| 56 | |
| 57 | def search_dripstack( |
| 58 | topic: str, |
| 59 | from_date: str | None = None, |
| 60 | to_date: str | None = None, |
| 61 | *, |
| 62 | depth: str = "default", |
| 63 | ) -> list[dict[str, Any]]: |
| 64 | """Search DripStack for articles matching the topic. |
| 65 | |
| 66 | Returns a list of raw item dicts from the search API. The free endpoint |
| 67 | requires no authentication. Results are relevance-ranked by DripStack's |
| 68 | own scoring (hybrid RRF — blended semantic + keyword match). |
| 69 | |
| 70 | Args: |
| 71 | topic: The search query (e.g. "AI capex risk", "Tesla earnings"). |
| 72 | from_date: ISO date string for start of window (YYYY-MM-DD). Not sent |
| 73 | to the API (DripStack search has its own time handling), but |
| 74 | available for post-filtering if needed. |
| 75 | to_date: ISO date string for end of window (YYYY-MM-DD). |
| 76 | depth: One of "quick", "default", "deep" — controls result count. |
| 77 | """ |
| 78 | limit = _DEPTH_LIMITS.get(depth, 10) |
| 79 | params = urllib.parse.urlencode({"q": topic, "limit": limit}) |
| 80 | url = f"{_SEARCH_URL}?{params}" |
| 81 | |
| 82 | try: |
| 83 | data = _get_json(url) |
| 84 | except Exception as e: |
| 85 | _log(f"search failed for '{topic}': {e}") |
| 86 | return [] |
| 87 | |
| 88 | items = data.get("items") or [] |
| 89 | if from_date or to_date: |
| 90 | windowed = [] |
| 91 | dropped = 0 |
| 92 | for item in items: |
| 93 | published = str(item.get("publishedAt") or "")[:10] |
| 94 | if published and from_date and published < from_date: |
| 95 | dropped += 1 |
| 96 | continue |
| 97 | if published and to_date and published > to_date: |
| 98 | dropped += 1 |
| 99 | continue |
| 100 | windowed.append(item) |
| 101 | if dropped: |
| 102 | _log(f"dropped {dropped} result(s) outside the {from_date}..{to_date} window") |
| 103 | items = windowed |
| 104 | _log(f"search '{topic}': {len(items)} results (confidence: {data.get('matchConfidence', '?')})") |
| 105 | return items |
| 106 | |
| 107 | |
| 108 | def parse_dripstack_response( |
| 109 | items: list[dict[str, Any]], |
| 110 | query: str = "", |
| 111 | ) -> list[dict[str, Any]]: |
| 112 | """Normalize DripStack search results into engine-style item dicts. |
| 113 | |
| 114 | Each item maps to the same shape as other sources (HN, Reddit, StockTwits): |
| 115 | id, title, url, author, date, engagement, relevance, why_relevant, |
| 116 | snippet, metadata. |
| 117 | |
| 118 | DripStack has no engagement signal (upvotes, likes), so engagement is |
| 119 | empty. Ranking relies on DripStack's own relevanceScore (0-100) which we |
| 120 | normalize to 0-1, plus recency. |
| 121 | """ |
| 122 | parsed: list[dict[str, Any]] = [] |
| 123 | for i, item in enumerate(items): |
| 124 | title = (item.get("title") or "").strip() |
| 125 | subtitle = (item.get("subtitle") or "").strip() |
| 126 | snippet_text = (item.get("snippet") or "").strip() |
| 127 | pub_slug = (item.get("publicationSlug") or "").strip() |
| 128 | post_slug = (item.get("slug") or "").strip() |
| 129 | published_at = (item.get("publishedAt") or "")[:10] or None |
| 130 | |
| 131 | # Build the article URL. For Substack-hosted publications the slug is |
| 132 | # the full hostname (e.g. "newsletter.doomberg.com") and the post slug |
| 133 | # is the path segment. For other domains the same pattern applies. |
| 134 | if pub_slug and post_slug: |
| 135 | url = f"https://{pub_slug}/{post_slug}" |
| 136 | else: |
| 137 | url = "" |
| 138 | |
| 139 | # Normalize DripStack's 0-100 relevanceScore to 0-1 for the engine. |
| 140 | raw_score = item.get("relevanceScore", 0) |
| 141 | try: |
| 142 | relevance = round(min(1.0, max(0.0, float(raw_score) / 100.0)), 2) |
| 143 | except (TypeError, ValueError): |
| 144 | relevance = 0.5 |
| 145 | |
| 146 | # Build a human-readable why_relevant from the whyMatched array. |
| 147 | why_parts = item.get("whyMatched") or [] |
| 148 | # Filter out internal RRF details; keep the useful match explanations. |
| 149 | why_clean = [ |
| 150 | w for w in why_parts |
| 151 | if "RRF" not in w and "Hybrid" not in w |
| 152 | ] |
| 153 | why_relevant = "; ".join(why_clean) if why_clean else f"DripStack newsletter match for: {query}" |
| 154 | |
| 155 | # The body feeds rerank and synthesis. Use subtitle (the article |
| 156 | # summary/lede) as the primary content, falling back to snippet. |
| 157 | body = subtitle or snippet_text or title |
| 158 | |
| 159 | # Publication name as author — gives attribution credit to the |
| 160 | # newsletter/analyst who wrote it (e.g. "SemiAnalysis", "Bloomberg"). |
| 161 | # Use the slug as a readable fallback. |
| 162 | author = pub_slug.replace(".substack.com", "").replace(".com", "") |
| 163 | |
| 164 | parsed.append({ |
| 165 | "id": f"DS{i + 1}", |
| 166 | "title": title or f"DripStack result {i + 1}", |
| 167 | "url": url, |
| 168 | "author": author or None, |
| 169 | "date": published_at, |
| 170 | "engagement": {}, |
| 171 | "relevance": relevance, |
| 172 | "why_relevant": why_relevant, |
| 173 | "body": body, |
| 174 | "snippet": snippet_text[:400], |
| 175 | "metadata": { |
| 176 | "publication_slug": pub_slug, |
| 177 | "post_slug": post_slug, |
| 178 | "relevance_score": raw_score, |
| 179 | "match_confidence": item.get("matchConfidence"), |
| 180 | "topic_coverage_ratio": item.get("topicCoverageRatio"), |
| 181 | }, |
| 182 | }) |
| 183 | return parsed |
| 184 | |
| 185 | |
| 186 | # --------------------------------------------------------------------------- # |
| 187 | # Standalone CLI # |
| 188 | # python3 dripstack.py "AI capex risk" # |
| 189 | # --------------------------------------------------------------------------- # |
| 190 | |
| 191 | if __name__ == "__main__": |
| 192 | topic = " ".join(sys.argv[1:]) or "AI capex" |
| 193 | today = datetime.date.today() |
| 194 | since = (today - datetime.timedelta(days=30)).isoformat() |
| 195 | |
| 196 | raw = search_dripstack(topic, from_date=since, depth="default") |
| 197 | items = parse_dripstack_response(raw, query=topic) |
| 198 | |
| 199 | print(f"Query: {topic} | {len(items)} results") |
| 200 | for it in items[:10]: |
| 201 | print(f" [{it['relevance']:.0%}] {it['title']} ({it['author']}, {it['date'] or 'no date'})") |
| 202 | if it["snippet"]: |
| 203 | print(f" {it['snippet'][:120]}") |
| 204 |