Google Docs · Equity Research Playbooknews, prices = market_api.snapshot(as_of=cutoff)
warehouse.append_raw(news, "RAW.MARKET.NEWS")
warehouse.append_raw(prices, "RAW.MARKET.DAILY_BARS")
RAW.MARKET
Build our stock-research pipeline. Land the market API’s news and daily prices in Snowflake, clean them with dbt, and use Python to extract LLM signals alongside our numerical trend model. Serve the results to BI through BigQuery.
I’ll inspect the ingestion job and Google Docs research brief first. News sentiment and numerical forecasts will keep separate provenance, then meet in a point-in-time research mart.
The sample has 12 article revisions. I’ll retain the latest version available at the research cutoff—not future revisions. LLM confidence describes extraction certainty, not a stock-price probability. Should low-confidence labels be excluded?
Keep them. Flag anything below 0.80 for review, and don’t lose events if the model fails. Run it in dev before we publish.
The dev rehearsal is ready: 228 unique events, with 211 ready for reporting, 15 flagged for review, and 2 retained with analysis errors. All seven dbt checks passed. The plan and notebook are open on the right; nothing has been published to production. BigQuery will receive separate semantic-signal and trend-forecast views for the research dashboard.
Ingest market data
news, prices = market_api.snapshot(as_of=cutoff)
warehouse.append_raw(news, "RAW.MARKET.NEWS")
warehouse.append_raw(prices, "RAW.MARKET.DAILY_BARS")
Prepare research inputs
dbt build --select int_news_by_ticker int_price_features
# Source URLs and as_of_at remain attached to every row.
Extract news signals
Structured output
- event
- Product launch
- sentiment
- Positive
- confidence
- 0.92
- evidence
- Source article retained
- review_status
- Ready
Model price trends
features = warehouse.table("INT_PRICE_FEATURES")
forecast = model.predict(features, horizon="5d")
forecast.save("RESEARCH.TREND_FORECASTS")
Build research mart
dbt build --select fct_stock_research
# Publishing requires approval; no production writes in this demo.
pipelines / researchenrich_news.py
from research.runtime import warehouse, llm, save_analysisfrom research.policy import review_status # Ingestion lands licensed API responses in Snowflake RAW.MARKET.# dbt prepares point-in-time news and numerical price features.news = warehouse.table("ANALYTICS.INTERMEDIATE.INT_NEWS_BY_TICKER") for batch in news.to_pandas_batches(): for row in batch.itertuples(): result = llm.extract( articles=row.ARTICLES, fields={"topic": str, "sentiment_score": float, "rationale": str, "confidence": float}, prompt_version="equity-v3", ) # The helper validates ranges and retains failed model calls. save_analysis( table="ANALYTICS.RESEARCH.NEWS_SIGNALS", security_id=row.SECURITY_ID, as_of_at=row.AS_OF_AT, result=result, review_status=review_status(result, threshold=0.80), preserve_sources=True, ) # predict_trends.py separately uses lagged returns and volume.# Its walk-forward forecast is not the LLM's confidence score.pipelines / researchenrich_news.py
from research.runtime import warehouse, llm, save_analysisfrom research.policy import review_status # Ingestion lands licensed API responses in Snowflake RAW.MARKET.# dbt prepares point-in-time news and numerical price features.news = warehouse.table("ANALYTICS.INTERMEDIATE.INT_NEWS_BY_TICKER") for batch in news.to_pandas_batches(): for row in batch.itertuples(): result = llm.extract( articles=row.ARTICLES, fields={"topic": str, "sentiment_score": float, "rationale": str, "confidence": float}, prompt_version="equity-v3", ) # The helper validates ranges and retains failed model calls. save_analysis( table="ANALYTICS.RESEARCH.NEWS_SIGNALS", security_id=row.SECURITY_ID, as_of_at=row.AS_OF_AT, result=result, review_status=review_status(result, threshold=0.80), preserve_sources=True, ) # predict_trends.py separately uses lagged returns and volume.# Its walk-forward forecast is not the LLM's confidence score.models / marts / researchfct_stock_research.sql
{{ config(materialized='table') }} with signals as ( select * from {{ source('research', 'news_signals') }}), forecasts as ( select * from {{ source('research', 'trend_forecasts') }}) select coalesce(s.security_id, f.security_id) as security_id, coalesce(s.as_of_at, f.as_of_at) as as_of_at, s.sentiment_score, s.confidence, s.review_status, s.source_urls, s.prompt_version, f.predicted_return, f.forecast_horizon, f.training_cutoff, f.model_versionfrom signals sfull outer join forecasts f on s.security_id = f.security_id and s.as_of_at = f.as_of_at -- Publish separate semantic and forecast views to BigQuery.-- Keep review states and numerical model provenance distinct.models / marts / researchfct_stock_research.yml
version: 2 sources: - name: research database: ANALYTICS schema: RESEARCH tables: - name: news_signals - name: trend_forecasts models: - name: fct_stock_research description: > Point-in-time equity research, served to BI through BigQuery. LLM confidence is not a probability of a stock-price move. data_tests: - dbt_utils.unique_combination_of_columns: arguments: combination_of_columns: [security_id, as_of_at] columns: - name: security_id data_tests: [not_null] - name: as_of_at data_tests: [not_null] - name: confidence data_tests: - dbt_utils.accepted_range: arguments: min_value: 0 max_value: 1 - name: sentiment_score data_tests: - dbt_utils.accepted_range: arguments: min_value: -1 max_value: 1