stocks-list.py

#!/usr/bin/env python3
"""
Stock price fetcher using Finnhub API and NewsAPI to populate a Slack List

This script fetches stock prices and news articles for a watchlist of stocks,
then creates a Slack List with the aggregated data. Specify which users you want
to share the list with in the USER_IDS variable and which stocks with the
STOCK_SYMBOLS variable.

Usage:
    python stocks-to-be-final.py

Environment Variables:
    SLACK_BOT_TOKEN: Create a Slack app https://api.slack.com/apps with Bot Token Scopes: lists:read, lists:write
    FINNHUB_API_KEY: API key from https://finnhub.io/
    NEWSAPI_API_KEY: API key from https://newsapi.org/

"""

import os
from datetime import datetime, timedelta

import finnhub
import newsapi
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

SLACK_BOT_TOKEN = os.getenv("SLACK_BOT_TOKEN")
FINNHUB_API_KEY = os.getenv("FINNHUB_API_KEY")
NEWSAPI_API_KEY = os.getenv("NEWSAPI_API_KEY")

NEWS_LOOKBACK_DAYS = 3
USER_IDS = [""]
LIST_NAME = "Jason's Watched Stocks"
LIST_DESCRIPTION = f"Script generated list of my watched stocks and the latest news"

LIST_SCHEMA = [
    {"key": "ticker_symbol", "name": "Ticker Symbol", "type": "text", "is_primary_column": True},
    {"key": "price", "name": "Price", "type": "number", "options": {"precision": 2}},
    {"key": "change", "name": "Change", "type": "number", "options": {"precision": 2}},
    {"key": "news", "name": "News", "type": "link"},
]

# Ticker symbols: search queries 
STOCK_SYMBOLS = {
    "CRM": "Salesforce",
    "MSFT": "Microsoft",
    "GOOG": "Alphabet",
    "AMZN": "Amazon",
    "NVDA": "NVIDIA"
}

slack_client = WebClient(token=SLACK_BOT_TOKEN)
finnhub_client = finnhub.Client(api_key=FINNHUB_API_KEY)
newsapi_client = newsapi.NewsApiClient(api_key=NEWSAPI_API_KEY)


def _make_slack_api_request(method, **kwargs):
    try:
        response = slack_client.api_call(method, json=kwargs)
        
        if not response.get('ok'):
            print(f"✗ Slack API error: {response.get('error', 'Unknown error')}")
            return None
            
        return response
    except SlackApiError as e:
        print(f"✗ Slack API error: {e.args[0]}")
        return None

def create_slack_list(list_name, list_description, list_schema):
    today_str = datetime.now().strftime("%m/%d/%Y %H:%M:%S")
    list_description = f"{list_description} as of {today_str}"
    
    response = _make_slack_api_request(
        "slackLists.create",    
        name=list_name,
        description_blocks=_build_rich_text_block(list_description),
        schema=list_schema,
    )
    
    list_id = response.get("list_id")
    full_schema = response.get("list_metadata").get("schema")
    
    return list_id, full_schema


def set_list_access(list_id, user_ids):
    _make_slack_api_request(
        "slackLists.access.set",
        list_id=list_id,
        user_ids=user_ids,
        access_level="write"
    )

def _build_rich_text_block(text):
    """Plain text columns must be `rich_text` within a table so that it matches the rest of Slack.
     More info: https://docs.slack.dev/reference/methods/slacklists.items.create#text-column-plain"""
    return [{"type": "rich_text", "elements": [{"type": "rich_text_section", "elements": [{"type": "text", "text": text}]}]}]


def get_stock_quote(symbol):
    quote = finnhub_client.quote(symbol)
    return quote


def get_top_article(query):
    from_date = (datetime.now() - timedelta(days=NEWS_LOOKBACK_DAYS)).strftime('%Y-%m-%d')
    news = newsapi_client.get_everything(q=query, page_size=1, sort_by='relevancy', language='en', from_param=from_date)
    return next((article.get('url') for article in news['articles'] if article.get('url')), "Nothing new")


def generate_initial_fields(symbol, quote, news, schema):
    return [
        {
            "column_id": schema[0].get("id"),
            "rich_text": _build_rich_text_block(symbol)
        },
        {
            "column_id": schema[1].get("id"),
            "number": [quote.get("c")]
        },
        {
            "column_id": schema[2].get("id"),
            "number": [quote.get("d")]
        },
        {
            "column_id": schema[3].get("id"),
            "link": [{"original_url": news}]
        }
    ]

def populate_list_items(list_id, full_schema):
    for symbol, search_query in STOCK_SYMBOLS.items():
        quote = get_stock_quote(symbol)
        news = get_top_article(search_query)
        
        initial_fields = generate_initial_fields(symbol, quote, news, full_schema)
        
        _make_slack_api_request(
            "slackLists.items.create",
            list_id=list_id,
            initial_fields=initial_fields
        )

def main():
    list_id, schema = create_slack_list(LIST_NAME, LIST_DESCRIPTION, LIST_SCHEMA)
    
    print(f"✓ Created Slack List: {list_id}")
    
    # Share the list with users (bot token lists are only visible to the bot by default)
    set_list_access(list_id, USER_IDS)
    print(f"✓ Set access for users: {USER_IDS}")
    
    populate_list_items(list_id, schema)
    print(f"✓ Populated list with {len(STOCK_SYMBOLS)} stocks")


if __name__ == "__main__":
    main()