c5251a05da
All stock price embeds now include a FinViz candlestick chart showing 5-minute intraday data with technical analysis indicators. Chart displays below price information in embed format. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
257 lines
8.6 KiB
Python
257 lines
8.6 KiB
Python
import discord
|
|
from discord.ext import commands
|
|
import logging
|
|
from datetime import datetime
|
|
from apscheduler.schedulers.asyncio import AsyncIOScheduler
|
|
from apscheduler.triggers.cron import CronTrigger
|
|
import pytz
|
|
|
|
from config import Config
|
|
from stock_api import YahooFinanceAPI, FinnhubAPI
|
|
from market_hours import MarketHours
|
|
|
|
|
|
# Configure logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StockBot(commands.Bot):
|
|
"""Discord bot for stock price tracking and reporting."""
|
|
|
|
def __init__(self):
|
|
intents = discord.Intents.default()
|
|
intents.message_content = True
|
|
super().__init__(command_prefix=Config.COMMAND_PREFIX, intents=intents)
|
|
|
|
# Initialize stock API based on configuration
|
|
if Config.STOCK_API_PROVIDER == 'finnhub':
|
|
self.stock_api = FinnhubAPI(Config.FINNHUB_API_KEY)
|
|
logger.info("Using Finnhub API for stock data")
|
|
else:
|
|
self.stock_api = YahooFinanceAPI()
|
|
logger.info("Using Yahoo Finance API for stock data")
|
|
|
|
self.scheduler = AsyncIOScheduler(timezone=pytz.timezone('America/New_York'))
|
|
self.target_channel_ids = [int(id) for id in Config.CHANNEL_IDS]
|
|
self.primary_ticker = Config.PRIMARY_TICKER
|
|
|
|
async def setup_hook(self):
|
|
"""Initialize the bot and set up scheduled tasks."""
|
|
logger.info("Bot setup initiated")
|
|
|
|
# Schedule hourly updates during market hours
|
|
# Run every hour on the hour, but only post if market is open
|
|
self.scheduler.add_job(
|
|
self.send_hourly_update,
|
|
CronTrigger(minute=0, timezone=MarketHours.NYSE_TIMEZONE),
|
|
id='hourly_update',
|
|
name='Hourly Stock Price Update'
|
|
)
|
|
|
|
# Schedule market close update at 4:00 PM ET on weekdays
|
|
self.scheduler.add_job(
|
|
self.send_market_close_update,
|
|
CronTrigger(hour=16, minute=0, day_of_week='mon-fri', timezone=MarketHours.NYSE_TIMEZONE),
|
|
id='market_close_update',
|
|
name='Market Close Update'
|
|
)
|
|
|
|
self.scheduler.start()
|
|
logger.info("Scheduler started for hourly updates and market close")
|
|
|
|
async def on_ready(self):
|
|
"""Called when bot successfully connects to Discord."""
|
|
logger.info(f'Bot connected as {self.user.name} (ID: {self.user.id})')
|
|
logger.info(f'Monitoring ticker: {self.primary_ticker}')
|
|
logger.info(f'Target channel IDs: {self.target_channel_ids}')
|
|
|
|
# Verify channels exist
|
|
for channel_id in self.target_channel_ids:
|
|
channel = self.get_channel(channel_id)
|
|
if channel:
|
|
logger.info(f'Target channel found: #{channel.name} (ID: {channel_id})')
|
|
else:
|
|
logger.error(f'Could not find channel with ID {channel_id}')
|
|
|
|
async def send_hourly_update(self):
|
|
"""Send hourly stock price update if market is open."""
|
|
if not MarketHours.is_market_open():
|
|
logger.info("Market is closed, skipping hourly update")
|
|
return
|
|
|
|
logger.info(f"Sending hourly update for {self.primary_ticker}")
|
|
for channel_id in self.target_channel_ids:
|
|
await self.send_stock_update(self.primary_ticker, channel_id)
|
|
|
|
async def send_market_close_update(self):
|
|
"""Send stock price update at market close on trading days."""
|
|
if not MarketHours.is_trading_day():
|
|
logger.info("Not a trading day, skipping market close update")
|
|
return
|
|
|
|
logger.info(f"Sending market close update for {self.primary_ticker}")
|
|
for channel_id in self.target_channel_ids:
|
|
await self.send_stock_update(self.primary_ticker, channel_id)
|
|
|
|
async def send_stock_update(self, ticker: str, channel_id: int):
|
|
"""
|
|
Fetch stock data and send formatted embed to specified channel.
|
|
|
|
Args:
|
|
ticker: Stock ticker symbol
|
|
channel_id: Discord channel ID to send message to
|
|
"""
|
|
channel = self.get_channel(channel_id)
|
|
if not channel:
|
|
logger.error(f"Could not find channel {channel_id}")
|
|
return
|
|
|
|
stock_data = self.stock_api.get_stock_price(ticker)
|
|
if not stock_data:
|
|
await channel.send(f"Unable to retrieve data for ticker: {ticker}")
|
|
return
|
|
|
|
embed = self.create_stock_embed(stock_data)
|
|
await channel.send(embed=embed)
|
|
logger.info(f"Sent stock update for {ticker} to channel {channel_id}")
|
|
|
|
def create_stock_embed(self, stock_data: dict) -> discord.Embed:
|
|
"""
|
|
Create a formatted Discord embed for stock price data.
|
|
|
|
Args:
|
|
stock_data: Dictionary containing stock information
|
|
|
|
Returns:
|
|
Discord embed object
|
|
"""
|
|
ticker = stock_data['ticker']
|
|
current_price = stock_data['current_price']
|
|
change_dollar = stock_data['change_dollar']
|
|
change_percent = stock_data['change_percent']
|
|
|
|
# Determine color based on price movement
|
|
if change_dollar > 0:
|
|
color = discord.Color.green()
|
|
change_emoji = "📈"
|
|
ansi_color = "\u001b[0;32m" # Green
|
|
elif change_dollar < 0:
|
|
color = discord.Color.red()
|
|
change_emoji = "📉"
|
|
ansi_color = "\u001b[0;31m" # Red
|
|
else:
|
|
color = discord.Color.blue()
|
|
change_emoji = "➡️"
|
|
ansi_color = "\u001b[0;34m" # Blue
|
|
|
|
# Format change string with ANSI color
|
|
change_sign = "+" if change_dollar >= 0 else ""
|
|
change_str = f"```ansi\n{ansi_color}{change_sign}${change_dollar} ({change_sign}{change_percent}%)\u001b[0m\n```"
|
|
|
|
# Custom title for PYPL ticker
|
|
if ticker == "PYPL":
|
|
title = f"{change_emoji} Steaks Stablecoin Value"
|
|
else:
|
|
title = f"{change_emoji} {ticker} Stock Price"
|
|
|
|
# Create embed
|
|
embed = discord.Embed(
|
|
title=title,
|
|
description=f"**${current_price}**",
|
|
color=color,
|
|
timestamp=datetime.now(pytz.timezone('America/New_York'))
|
|
)
|
|
|
|
embed.add_field(name="Change", value=change_str, inline=True)
|
|
embed.add_field(name="Previous Close", value=f"${stock_data['previous_close']}", inline=True)
|
|
|
|
# Add FinViz 5-minute intraday chart
|
|
chart_url = f"https://finviz.com/chart.ashx?t={ticker}&ty=c&ta=1&p=i5&s=l"
|
|
embed.set_image(url=chart_url)
|
|
|
|
market_status = "🟢 Market Open" if stock_data['market_open'] else "🔴 Market Closed"
|
|
embed.set_footer(text=f"{market_status}")
|
|
|
|
return embed
|
|
|
|
|
|
# Create bot instance
|
|
bot = StockBot()
|
|
|
|
|
|
@bot.command(name='stock', aliases=['price'])
|
|
async def get_stock_price(ctx, ticker: str = None):
|
|
"""
|
|
Manually query stock price for any ticker.
|
|
|
|
Usage: !stock AAPL or !price TSLA
|
|
"""
|
|
if not ticker:
|
|
await ctx.send("Please provide a ticker symbol. Usage: `!stock AAPL`")
|
|
return
|
|
|
|
ticker = ticker.upper()
|
|
logger.info(f"Manual stock query for {ticker} by {ctx.author}")
|
|
|
|
stock_data = bot.stock_api.get_stock_price(ticker)
|
|
if not stock_data:
|
|
await ctx.send(f"Unable to retrieve data for ticker: {ticker}")
|
|
return
|
|
|
|
embed = bot.create_stock_embed(stock_data)
|
|
await ctx.send(embed=embed)
|
|
|
|
|
|
@bot.command(name='ping')
|
|
async def ping(ctx):
|
|
"""Check if bot is responsive."""
|
|
await ctx.send(f'Pong! Latency: {round(bot.latency * 1000)}ms')
|
|
|
|
|
|
@bot.command(name='market')
|
|
async def market_status(ctx):
|
|
"""Check if market is currently open."""
|
|
is_open = MarketHours.is_market_open()
|
|
status = "🟢 Market is currently OPEN" if is_open else "🔴 Market is currently CLOSED"
|
|
|
|
embed = discord.Embed(
|
|
title="NYSE Market Status",
|
|
description=status,
|
|
color=discord.Color.green() if is_open else discord.Color.red(),
|
|
timestamp=datetime.now(pytz.timezone('America/New_York'))
|
|
)
|
|
|
|
embed.add_field(
|
|
name="Trading Hours",
|
|
value="Monday-Friday\n9:30 AM - 4:00 PM ET",
|
|
inline=False
|
|
)
|
|
|
|
if not is_open:
|
|
next_open = MarketHours.get_next_market_open()
|
|
embed.add_field(
|
|
name="Next Open",
|
|
value=next_open.strftime("%A, %B %d at %I:%M %p ET"),
|
|
inline=False
|
|
)
|
|
|
|
await ctx.send(embed=embed)
|
|
|
|
|
|
def main():
|
|
"""Main entry point for the bot."""
|
|
if not Config.validate():
|
|
logger.error("Configuration validation failed")
|
|
return
|
|
|
|
logger.info("Starting bot...")
|
|
bot.run(Config.DISCORD_TOKEN)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|