X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=stock_renderer.py;h=b13af908235908428ea1e0dc747fca0424624058;hb=ea1ee5f817c01c3736a64d73d496cf35cbd383e5;hp=f8491e6a9eb73b8ba06ea35ffd1eb96f5ca0b5c2;hpb=75b27cc68871343681f01e3f5b04cae84b1b7b2a;p=kiosk.git diff --git a/stock_renderer.py b/stock_renderer.py index f8491e6..b13af90 100644 --- a/stock_renderer.py +++ b/stock_renderer.py @@ -1,129 +1,209 @@ -from bs4 import BeautifulSoup -from threading import Thread -import datetime +#!/usr/bin/env python3 + +import logging +import os +from typing import Any, Dict, List, Optional, Tuple + +import yfinance as yf # type: ignore +import yahooquery as yq # type: ignore +import plotly.graph_objects as go + import file_writer -import json -import re +import kiosk_constants import renderer -import random -import secrets -import time -import urllib.request, urllib.error, urllib.parse - -class stock_quote_renderer(renderer.debuggable_abstaining_renderer): - # format exchange:symbol - def __init__(self, name_to_timeout_dict, symbols): - super(stock_quote_renderer, self).__init__(name_to_timeout_dict, False) - self.symbols = symbols - self.prefix = "https://www.alphavantage.co/query?" - self.thread = None - - def debug_prefix(self): - return "stock" - - def get_random_key(self): - return random.choice(secrets.alphavantage_keys) - - def periodic_render(self, key): - now = datetime.datetime.now() - if (now.hour < (9 - 3) or - now.hour >= (17 - 3) or - datetime.datetime.today().weekday() > 4): - self.debug_print("The stock market is closed so not re-rendering") - return True - - if (self.thread is None or not self.thread.is_alive()): - self.debug_print("Spinning up a background thread...") - self.thread = Thread(target = self.thread_internal_render, args=()) - self.thread.start() - return True - def thread_internal_render(self): - symbols_finished = 0 - f = file_writer.file_writer('stock_3_86400.html') - f.write("
+
+class stock_quote_renderer(renderer.abstaining_renderer):
+ """Render the stock prices page."""
+
+ def __init__(
+ self,
+ name_to_timeout_dict: Dict[str, int],
+ symbols: List[str],
+ display_subs: Dict[str, str] = None,
+ ) -> None:
+ super().__init__(name_to_timeout_dict)
+ self.backend = "yahooquery"
+ self.symbols = symbols
+ self.display_subs = display_subs
+ self.info_cache: Dict[str, Any] = {}
+
+ def get_financial_data(self, symbol: str) -> Optional[Any]:
+ if symbol in self.info_cache:
+ return self.info_cache[symbol]
+ if self.backend == "yahooquery":
+ ticker = yq.Ticker(symbol)
+ if "Quote not found" in ticker.price[symbol]:
+ return None
+ self.info_cache[symbol] = ticker
+ return ticker
+ elif self.backend == "yfinance":
+ ticker = yf.Ticker(symbol)
+ if not ticker.fast_info["last_price"]:
+ return None
+ self.info_cache[symbol] = ticker
+ return ticker
+ else:
+ raise Exception(f"Unknown backend: {self.backend}")
+
+ def get_ticker_name(self, ticker: Any) -> Optional[str]:
+ """Get friendly name of a ticker."""
+ if isinstance(ticker, yf.Ticker):
+ return ticker.ticker
+ elif isinstance(ticker, yq.Ticker):
+ return ticker.symbols[0]
+ return None
+
+ @staticmethod
+ def prioritized_get_item_from_dict(
+ keys: List[str], dictionary: Dict[str, Any]
+ ) -> Optional[Any]:
+ result = None
+ for key in keys:
+ result = dictionary.get(key, None)
+ if result:
+ return result
+ return None
+
+ def get_price(self, ticker: Any) -> Optional[float]:
+ """Get most recent price of a ticker."""
+ if isinstance(ticker, yf.Ticker):
+ price = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["last_price", "open", "previous_close"], ticker.fast_info
+ )
+ if price:
+ return price
+
+ price = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["bid", "ask", "lastMarket"],
+ ticker.info,
+ )
+ if price:
+ return price
+ return None
+ elif isinstance(ticker, yq.Ticker):
+ price = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["regularMarketPrice", "preMarketPrice", "regularMarketPreviousClose"],
+ ticker.price[ticker.symbols[0]],
+ )
+ if price:
+ return price
+ return None
+
+ def get_last_close(self, ticker: Any) -> Optional[float]:
+ if isinstance(ticker, yf.Ticker):
+ last_close = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["previous_close", "open"], ticker.fast_info
+ )
+ if last_close:
+ return last_close
+
+ last_close = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["preMarketPrice"], ticker.info
+ )
+ if last_close:
+ return last_close
+ elif isinstance(ticker, yq.Ticker):
+ last_close = stock_quote_renderer.prioritized_get_item_from_dict(
+ ["regularMarketPreviousClose", "regularMarketOpen"],
+ ticker.price[ticker.symbols[0]],
+ )
+ if last_close:
+ return last_close
+ return self.get_price(ticker)
+
+ def get_change_and_delta(
+ self, ticker: Any, current_price: float
+ ) -> Tuple[float, float]:
+ """Given the current price, look up opening price and compute delta."""
+ last_price = self.get_last_close(ticker)
+ delta = current_price - last_price
+ return (delta / last_price * 100.0, delta)
+
+ @staticmethod
+ def make_chart(symbol: str, ticker: Any, period: str) -> str:
+ base_filename = f"stock_chart_{symbol}.png"
+ output_filename = os.path.join(kiosk_constants.pages_dir, base_filename)
+ transparent = go.Layout(
+ paper_bgcolor="rgba(0,0,0,0)",
+ plot_bgcolor="rgba(0,0,0,0)",
+ xaxis_rangeslider_visible=False,
+ )
+
+ hist = ticker.history(period=period, interval="1wk")
+ if isinstance(ticker, yq.Ticker):
+ _open = "open"
+ _high = "high"
+ _low = "low"
+ _close = "adjclose"
+ elif isinstance(ticker, yf.Ticker):
+ _open = "Open"
+ _high = "High"
+ _low = "Low"
+ _close = "Close"
+ else:
+ raise Exception("Bad Ticker type")
+ chart = go.Figure(
+ data=go.Candlestick(
+ open=hist[_open],
+ high=hist[_high],
+ low=hist[_low],
+ close=hist[_close],
+ ),
+ layout=transparent,
+ )
+ chart.update_xaxes(visible=False, showticklabels=False)
+ chart.update_yaxes(side="right")
+ chart.write_image(output_filename, format="png", width=600, height=350)
+ print(f"Write {output_filename}...")
+ return base_filename
+
+ def periodic_render(self, key: str) -> bool:
+ """Write an up-to-date stock page."""
+ with file_writer.file_writer("stock_3_86400.html") as f:
+ f.write("Stock Quotes") + f.write("
| """
+ )
+ f.write("