Anson/Simple-Stock-Bot
1
0
Fork 0
mirror of https://gitlab.com/simple-stock-bots/simple-stock-bot.git synced 2025-06-16 07:16:40 +00:00
Code Releases Activity

Merge branch 'canary' into 'master'

Browse Source 
Trending Update

Closes #73 and #74

See merge request simple-stock-bots/simple-telegram-stock-bot!29
This commit is contained in:
Anson Biggs 2021-09-01 03:23:01 +00:00
commit 88a9b3aa63
7 changed files with 1447 additions and 1340 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
DockerDev → .devcontainer/Dockerfile
View File

2
.devcontainer/devcontainer.json
View File

@ -5,7 +5,7 @@
// Sets the run context to one level up instead of the .devcontainer folder. // Sets the run context to one level up instead of the .devcontainer folder.
"context": "..", "context": "..",
// Update the 'dockerFile' property if you aren't using the standard 'Dockerfile' filename. // Update the 'dockerFile' property if you aren't using the standard 'Dockerfile' filename.
"dockerFile": "../DockerDev", "dockerFile": "Dockerfile",
// Set *default* container specific settings.json values on container create. // Set *default* container specific settings.json values on container create.
"settings": {}, "settings": {},
// Add the IDs of extensions you want installed when the container is created. // Add the IDs of extensions you want installed when the container is created.

1022
IEX_Symbol.py
View File

File diff suppressed because it is too large Load Diff

4
Symbol.py
View File

@ -8,6 +8,7 @@ class Symbol:
symbol: What the user calls it. ie tsla or btc symbol: What the user calls it. ie tsla or btc
id: What the api expects. ie tsla or bitcoin id: What the api expects. ie tsla or bitcoin
name: Human readable. ie Tesla or Bitcoin name: Human readable. ie Tesla or Bitcoin
tag: Uppercase tag to call the symbol. ie $TSLA or $$BTC
""" """
currency = "usd" currency = "usd"
@ -17,6 +18,7 @@ class Symbol:
self.symbol = symbol self.symbol = symbol
self.id = symbol self.id = symbol
self.name = symbol self.name = symbol
self.tag = "$" + symbol
def __repr__(self) -> str: def __repr__(self) -> str:
return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>" return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>"
@ -32,6 +34,7 @@ class Stock(Symbol):
self.symbol = symbol self.symbol = symbol
self.id = symbol self.id = symbol
self.name = "$" + symbol.upper() self.name = "$" + symbol.upper()
self.tag = "$" + symbol.upper()
# Used by Coin to change symbols for ids # Used by Coin to change symbols for ids
@ -44,6 +47,7 @@ class Coin(Symbol):
@functools.cache @functools.cache
def __init__(self, symbol: str) -> None: def __init__(self, symbol: str) -> None:
self.symbol = symbol self.symbol = symbol
self.tag = "$$" + symbol.upper()
self.get_data() self.get_data()
def get_data(self) -> None: def get_data(self) -> None:

6
bot.py
View File

@ -116,12 +116,12 @@ def donate(update: Update, context: CallbackContext):
except ValueError: except ValueError:
update.message.reply_text(f"{amount} is not a valid donation amount or number.") update.message.reply_text(f"{amount} is not a valid donation amount or number.")
return return
info(f"Donation amount: {price}") info(f"Donation amount: {price} by {update.message.chat.username}")
context.bot.send_invoice( context.bot.send_invoice(
chat_id=chat_id, chat_id=chat_id,
title="Simple Stock Bot Donation", title="Simple Stock Bot Donation",
description=f"Simple Stock Bot Donation of ${amount}", description=f"Simple Stock Bot Donation of ${amount} by {update.message.chat.username}",
payload=f"simple-stock-bot-{chat_id}", payload=f"simple-stock-bot-{chat_id}",
provider_token=STRIPE_TOKEN, provider_token=STRIPE_TOKEN,
currency="USD", currency="USD",
@ -560,7 +560,7 @@ def main():
dp.add_handler(CommandHandler("stat", stat)) dp.add_handler(CommandHandler("stat", stat))
dp.add_handler(CommandHandler("stats", stat)) dp.add_handler(CommandHandler("stats", stat))
dp.add_handler(CommandHandler("cap", cap)) dp.add_handler(CommandHandler("cap", cap))
dp.add_handler(CommandHandler("trending", trending)) dp.add_handler(CommandHandler("trending", trending, run_async=True))
dp.add_handler(CommandHandler("search", search)) dp.add_handler(CommandHandler("search", search))
dp.add_handler(CommandHandler("random", rand_pick)) dp.add_handler(CommandHandler("random", rand_pick))
dp.add_handler(CommandHandler("donate", donate)) dp.add_handler(CommandHandler("donate", donate))

835
cg_Crypto.py
View File

@ -1,410 +1,425 @@
"""Class with functions for running the bot with IEX Cloud. """Class with functions for running the bot with IEX Cloud.
""" """
import logging import logging
from datetime import datetime from datetime import datetime
from typing import List, Optional, Tuple from logging import critical, debug, error, info, warning
from typing import List, Optional, Tuple
import pandas as pd
import requests as r import pandas as pd
import schedule import requests as r
from fuzzywuzzy import fuzz import schedule
from markdownify import markdownify from fuzzywuzzy import fuzz
from markdownify import markdownify
from Symbol import Coin
from Symbol import Coin
class cg_Crypto:
""" class cg_Crypto:
Functions for finding crypto info """
""" Functions for finding crypto info
"""
vs_currency = "usd" # simple/supported_vs_currencies for list of options
vs_currency = "usd" # simple/supported_vs_currencies for list of options
searched_symbols = {}
searched_symbols = {}
def __init__(self) -> None: trending_cache = None
"""Creates a Symbol Object
def __init__(self) -> None:
Parameters """Creates a Symbol Object
----------
IEX_TOKEN : str Parameters
IEX Token ----------
""" IEX_TOKEN : str
self.get_symbol_list() IEX Token
schedule.every().day.do(self.get_symbol_list) """
self.get_symbol_list()
def get(self, endpoint, params: dict = {}, timeout=10) -> dict: schedule.every().day.do(self.get_symbol_list)
url = "https://api.coingecko.com/api/v3" + endpoint def get(self, endpoint, params: dict = {}, timeout=10) -> dict:
resp = r.get(url, params=params, timeout=timeout)
# Make sure API returned a proper status code url = "https://api.coingecko.com/api/v3" + endpoint
try: resp = r.get(url, params=params, timeout=timeout)
resp.raise_for_status() # Make sure API returned a proper status code
except r.exceptions.HTTPError as e: try:
logging.error(e) resp.raise_for_status()
return {} except r.exceptions.HTTPError as e:
logging.error(e)
# Make sure API returned valid JSON return {}
try:
resp_json = resp.json() # Make sure API returned valid JSON
return resp_json try:
except r.exceptions.JSONDecodeError as e: resp_json = resp.json()
logging.error(e) return resp_json
return {} except r.exceptions.JSONDecodeError as e:
logging.error(e)
def symbol_id(self, symbol) -> str: return {}
try:
return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[ def symbol_id(self, symbol) -> str:
0 try:
] return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[
except KeyError: 0
return "" ]
except KeyError:
def get_symbol_list( return ""
self, return_df=False
) -> Optional[Tuple[pd.DataFrame, datetime]]: def get_symbol_list(
self, return_df=False
raw_symbols = self.get("/coins/list") ) -> Optional[Tuple[pd.DataFrame, datetime]]:
symbols = pd.DataFrame(data=raw_symbols)
raw_symbols = self.get("/coins/list")
symbols["description"] = ( symbols = pd.DataFrame(data=raw_symbols)
"$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
) symbols["description"] = (
symbols = symbols[["id", "symbol", "name", "description"]] "$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
symbols["type_id"] = "$$" + symbols["id"] )
symbols = symbols[["id", "symbol", "name", "description"]]
self.symbol_list = symbols symbols["type_id"] = "$$" + symbols["id"]
if return_df:
return symbols, datetime.now() self.symbol_list = symbols
if return_df:
def status(self) -> str: return symbols, datetime.now()
"""Checks CoinGecko /ping endpoint for API issues.
def status(self) -> str:
Returns """Checks CoinGecko /ping endpoint for API issues.
-------
str Returns
Human readable text on status of CoinGecko API -------
""" str
status = r.get( Human readable text on status of CoinGecko API
"https://api.coingecko.com/api/v3/ping", """
timeout=5, status = r.get(
) "https://api.coingecko.com/api/v3/ping",
timeout=5,
try: )
status.raise_for_status()
return f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds." try:
except: status.raise_for_status()
return f"CoinGecko API returned an error code {status.status_code} in {status.elapsed.total_seconds()} Seconds." return f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds."
except:
def search_symbols(self, search: str) -> List[Tuple[str, str]]: return f"CoinGecko API returned an error code {status.status_code} in {status.elapsed.total_seconds()} Seconds."
"""Performs a fuzzy search to find coin symbols closest to a search term.
def search_symbols(self, search: str) -> List[Tuple[str, str]]:
Parameters """Performs a fuzzy search to find coin symbols closest to a search term.
----------
search : str Parameters
String used to search, could be a company name or something close to the companies coin ticker. ----------
search : str
Returns String used to search, could be a company name or something close to the companies coin ticker.
-------
List[tuple[str, str]] Returns
A list tuples of every coin sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name). -------
""" List[tuple[str, str]]
schedule.run_pending() A list tuples of every coin sorted in order of how well they match. Each tuple contains: (Symbol, Issue Name).
search = search.lower() """
try: # https://stackoverflow.com/a/3845776/8774114 schedule.run_pending()
return self.searched_symbols[search] search = search.lower()
except KeyError: try: # https://stackoverflow.com/a/3845776/8774114
pass return self.searched_symbols[search]
except KeyError:
symbols = self.symbol_list pass
symbols["Match"] = symbols.apply(
lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()), symbols = self.symbol_list
axis=1, symbols["Match"] = symbols.apply(
) lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
axis=1,
symbols.sort_values(by="Match", ascending=False, inplace=True) )
if symbols["Match"].head().sum() < 300:
symbols["Match"] = symbols.apply( symbols.sort_values(by="Match", ascending=False, inplace=True)
lambda x: fuzz.partial_ratio(search, x["name"].lower()), if symbols["Match"].head().sum() < 300:
axis=1, symbols["Match"] = symbols.apply(
) lambda x: fuzz.partial_ratio(search, x["name"].lower()),
axis=1,
symbols.sort_values(by="Match", ascending=False, inplace=True) )
symbols = symbols.head(10)
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"]))) symbols.sort_values(by="Match", ascending=False, inplace=True)
self.searched_symbols[search] = symbol_list symbols = symbols.head(10)
return symbol_list symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
self.searched_symbols[search] = symbol_list
def price_reply(self, coin: Coin) -> str: return symbol_list
"""Returns current market price or after hours if its available for a given coin symbol.
def price_reply(self, coin: Coin) -> str:
Parameters """Returns current market price or after hours if its available for a given coin symbol.
----------
symbols : list Parameters
List of coin symbols. ----------
symbols : list
Returns List of coin symbols.
-------
Dict[str, str] Returns
Each symbol passed in is a key with its value being a human readable -------
markdown formatted string of the symbols price and movement. Dict[str, str]
""" Each symbol passed in is a key with its value being a human readable
markdown formatted string of the symbols price and movement.
if resp := self.get( """
"/simple/price",
params={ if resp := self.get(
"ids": coin.id, "/simple/price",
"vs_currencies": self.vs_currency, params={
"include_24hr_change": "true", "ids": coin.id,
}, "vs_currencies": self.vs_currency,
): "include_24hr_change": "true",
try: },
data = resp[coin.id] ):
try:
price = data[self.vs_currency] data = resp[coin.id]
change = data[self.vs_currency + "_24h_change"]
if change is None: price = data[self.vs_currency]
change = 0 change = data[self.vs_currency + "_24h_change"]
except KeyError: if change is None:
return f"{coin.id} returned an error." change = 0
except KeyError:
message = f"The current price of {coin.name} is $**{price:,}**" return f"{coin.id} returned an error."
# Determine wording of change text message = f"The current price of {coin.name} is $**{price:,}**"
if change > 0:
message += f", the coin is currently **up {change:.3f}%** for today" # Determine wording of change text
elif change < 0: if change > 0:
message += f", the coin is currently **down {change:.3f}%** for today" message += f", the coin is currently **up {change:.3f}%** for today"
else: elif change < 0:
message += ", the coin hasn't shown any movement today." message += f", the coin is currently **down {change:.3f}%** for today"
else:
else: message += ", the coin hasn't shown any movement today."
message = f"The price for {coin.name} is not available. If you suspect this is an error run `/status`"
else:
return message message = f"The price for {coin.name} is not available. If you suspect this is an error run `/status`"
def intra_reply(self, symbol: Coin) -> pd.DataFrame: return message
"""Returns price data for a symbol since the last market open.
def intra_reply(self, symbol: Coin) -> pd.DataFrame:
Parameters """Returns price data for a symbol since the last market open.
----------
symbol : str Parameters
Stock symbol. ----------
symbol : str
Returns Stock symbol.
-------
pd.DataFrame Returns
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. -------
""" pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
if resp := self.get( """
f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 1}, if resp := self.get(
): f"/coins/{symbol.id}/ohlc",
df = pd.DataFrame( params={"vs_currency": self.vs_currency, "days": 1},
resp, columns=["Date", "Open", "High", "Low", "Close"] ):
).dropna() df = pd.DataFrame(
df["Date"] = pd.to_datetime(df["Date"], unit="ms") resp, columns=["Date", "Open", "High", "Low", "Close"]
df = df.set_index("Date") ).dropna()
return df df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date")
return pd.DataFrame() return df
def chart_reply(self, symbol: Coin) -> pd.DataFrame: return pd.DataFrame()
"""Returns price data for a symbol of the past month up until the previous trading days close.
Also caches multiple requests made in the same day. def chart_reply(self, symbol: Coin) -> pd.DataFrame:
"""Returns price data for a symbol of the past month up until the previous trading days close.
Parameters Also caches multiple requests made in the same day.
----------
symbol : str Parameters
Stock symbol. ----------
symbol : str
Returns Stock symbol.
-------
pd.DataFrame Returns
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. -------
""" pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
if resp := self.get( """
f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 30}, if resp := self.get(
): f"/coins/{symbol.id}/ohlc",
df = pd.DataFrame( params={"vs_currency": self.vs_currency, "days": 30},
resp, columns=["Date", "Open", "High", "Low", "Close"] ):
).dropna() df = pd.DataFrame(
df["Date"] = pd.to_datetime(df["Date"], unit="ms") resp, columns=["Date", "Open", "High", "Low", "Close"]
df = df.set_index("Date") ).dropna()
return df df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date")
return pd.DataFrame() return df
def stat_reply(self, symbol: Coin) -> str: return pd.DataFrame()
"""Gathers key statistics on coin. Mostly just CoinGecko scores.
def stat_reply(self, symbol: Coin) -> str:
Parameters """Gathers key statistics on coin. Mostly just CoinGecko scores.
----------
symbol : Coin Parameters
----------
Returns symbol : Coin
-------
str Returns
Preformatted markdown. -------
""" str
Preformatted markdown.
if data := self.get( """
f"/coins/{symbol.id}",
params={ if data := self.get(
"localization": "false", f"/coins/{symbol.id}",
}, params={
): "localization": "false",
},
return f""" ):
[{data['name']}]({data['links']['homepage'][0]}) Statistics:
Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,} return f"""
Market Cap Ranking: {data.get('market_cap_rank',"Not Available")} [{data['name']}]({data['links']['homepage'][0]}) Statistics:
CoinGecko Scores: Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,}
Overall: {data.get('coingecko_score','Not Available')} Market Cap Ranking: {data.get('market_cap_rank',"Not Available")}
Development: {data.get('developer_score','Not Available')} CoinGecko Scores:
Community: {data.get('community_score','Not Available')} Overall: {data.get('coingecko_score','Not Available')}
Public Interest: {data.get('public_interest_score','Not Available')} Development: {data.get('developer_score','Not Available')}
""" Community: {data.get('community_score','Not Available')}
else: Public Interest: {data.get('public_interest_score','Not Available')}
return f"{symbol.symbol} returned an error." """
else:
def cap_reply(self, coin: Coin) -> str: return f"{symbol.symbol} returned an error."
"""Gets market cap for Coin
def cap_reply(self, coin: Coin) -> str:
Parameters """Gets market cap for Coin
----------
coin : Coin Parameters
----------
Returns coin : Coin
-------
str Returns
Preformatted markdown. -------
""" str
Preformatted markdown.
if resp := self.get( """
f"/simple/price",
params={ if resp := self.get(
"ids": coin.id, f"/simple/price",
"vs_currencies": self.vs_currency, params={
"include_market_cap": "true", "ids": coin.id,
}, "vs_currencies": self.vs_currency,
): "include_market_cap": "true",
print(resp) },
try: ):
data = resp[coin.id] debug(resp)
try:
price = data[self.vs_currency] data = resp[coin.id]
cap = data[self.vs_currency + "_market_cap"]
except KeyError: price = data[self.vs_currency]
return f"{coin.id} returned an error." cap = data[self.vs_currency + "_market_cap"]
except KeyError:
if cap == 0: return f"{coin.id} returned an error."
return f"The market cap for {coin.name} is not available for unknown reasons."
if cap == 0:
message = f"The current price of {coin.name} is $**{price:,}** and its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}" return f"The market cap for {coin.name} is not available for unknown reasons."
else: message = f"The current price of {coin.name} is $**{price:,}** and its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}"
message = f"The Coin: {coin.name} was not found or returned and error."
else:
return message message = f"The Coin: {coin.name} was not found or returned and error."
def info_reply(self, symbol: Coin) -> str: return message
"""Gets coin description
def info_reply(self, symbol: Coin) -> str:
Parameters """Gets coin description
----------
symbol : Coin Parameters
----------
Returns symbol : Coin
-------
str Returns
Preformatted markdown. -------
""" str
Preformatted markdown.
if data := self.get( """
f"/coins/{symbol.id}",
params={"localization": "false"}, if data := self.get(
): f"/coins/{symbol.id}",
try: params={"localization": "false"},
return markdownify(data["description"]["en"]) ):
except KeyError: try:
return f"{symbol} does not have a description available." return markdownify(data["description"]["en"])
except KeyError:
return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist." return f"{symbol} does not have a description available."
def trending(self) -> list[str]: return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
"""Gets current coins trending on coingecko
def spark_reply(self, symbol: Coin) -> str:
Returns change = self.get(
------- f"/simple/price",
list[str] params={
list of $$ID: NAME, CHANGE% "ids": symbol.id,
""" "vs_currencies": self.vs_currency,
"include_24hr_change": "true",
coins = self.get("/search/trending") },
try: )[symbol.id]["usd_24h_change"]
trending = []
for coin in coins["coins"]: return f"`{symbol.tag}`: {symbol.name}, {change:.2f}%"
c = coin["item"]
def trending(self) -> list[str]:
sym = c["symbol"].upper() """Gets current coins trending on coingecko
name = c["name"]
change = self.get( Returns
f"/simple/price", -------
params={ list[str]
"ids": c["id"], list of $$ID: NAME, CHANGE%
"vs_currencies": self.vs_currency, """
"include_24hr_change": "true",
}, coins = self.get("/search/trending")
)[c["id"]]["usd_24h_change"] try:
trending = []
msg = f"`$${sym}`: {name}, {change:.2f}%" for coin in coins["coins"]:
c = coin["item"]
trending.append(msg)
sym = c["symbol"].upper()
except Exception as e: name = c["name"]
logging.warning(e) change = self.get(
trending = ["Trending Coins Currently Unavailable."] f"/simple/price",
params={
return trending "ids": c["id"],
"vs_currencies": self.vs_currency,
def batch_price(self, coins: list[Coin]) -> list[str]: "include_24hr_change": "true",
"""Gets price of a list of coins all in one API call },
)[c["id"]]["usd_24h_change"]
Parameters
---------- msg = f"`$${sym}`: {name}, {change:.2f}%"
coins : list[Coin]
trending.append(msg)
Returns
------- except Exception as e:
list[str] logging.warning(e)
returns preformatted list of strings detailing price movement of each coin passed in. return self.trending_cache
"""
query = ",".join([c.id for c in coins]) self.trending_cache = trending
return trending
prices = self.get(
f"/simple/price", def batch_price(self, coins: list[Coin]) -> list[str]:
params={ """Gets price of a list of coins all in one API call
"ids": query,
"vs_currencies": self.vs_currency, Parameters
"include_24hr_change": "true", ----------
}, coins : list[Coin]
)
Returns
replies = [] -------
for coin in coins: list[str]
if coin.id in prices: returns preformatted list of strings detailing price movement of each coin passed in.
p = prices[coin.id] """
query = ",".join([c.id for c in coins])
if p.get("usd_24h_change") is None:
p["usd_24h_change"] = 0 prices = self.get(
f"/simple/price",
replies.append( params={
f"{coin.name}: ${p.get('usd',0):,} and has moved {p.get('usd_24h_change',0.0):.2f}% in the past 24 hours." "ids": query,
) "vs_currencies": self.vs_currency,
"include_24hr_change": "true",
return replies },
)
replies = []
for coin in coins:
if coin.id in prices:
p = prices[coin.id]
if p.get("usd_24h_change") is None:
p["usd_24h_change"] = 0
replies.append(
f"{coin.name}: ${p.get('usd',0):,} and has moved {p.get('usd_24h_change',0.0):.2f}% in the past 24 hours."
)
return replies

918
symbol_router.py
View File

@ -1,424 +1,494 @@
"""Function that routes symbols to the correct API provider. """Function that routes symbols to the correct API provider.
""" """
import datetime import datetime
import random import random
import re import re
from logging import critical, debug, error, info, warning from logging import critical, debug, error, info, warning
import pandas as pd import pandas as pd
from fuzzywuzzy import fuzz import schedule
from fuzzywuzzy import fuzz
from cg_Crypto import cg_Crypto
from IEX_Symbol import IEX_Symbol from cg_Crypto import cg_Crypto
from Symbol import Coin, Stock, Symbol from IEX_Symbol import IEX_Symbol
from Symbol import Coin, Stock, Symbol
class Router:
STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z.]{1,6})" class Router:
CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,20})" STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z.]{1,6})"
searched_symbols = {} CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,20})"
searched_symbols = {}
def __init__(self): trending_count = {}
self.stock = IEX_Symbol()
self.crypto = cg_Crypto() def __init__(self):
self.stock = IEX_Symbol()
def find_symbols(self, text: str) -> list[Symbol]: self.crypto = cg_Crypto()
"""Finds stock tickers starting with a dollar sign, and cryptocurrencies with two dollar signs
in a blob of text and returns them in a list. schedule.every().hour.do(self.trending_decay)
Parameters def trending_decay(self, decay=0.5):
---------- """Decays the value of each trending stock by a multiplier"""
text : str
Blob of text. info("Decaying trending symbols.")
if self.trending_count:
Returns for key in self.trending_count.keys():
------- if self.trending_count[key] < 0.01:
list[str] # This just makes sure were not keeping around keys that havent been called in a very long time.
List of stock symbols as strings without dollar sign. self.trending_count.pop(key, None)
""" else:
symbols = [] self.trending_count[key] = self.trending_count[key] * decay
stocks = set(re.findall(self.STOCK_REGEX, text))
for stock in stocks: def find_symbols(self, text: str) -> list[Symbol]:
if stock.upper() in self.stock.symbol_list["symbol"].values: """Finds stock tickers starting with a dollar sign, and cryptocurrencies with two dollar signs
symbols.append(Stock(stock)) in a blob of text and returns them in a list.
else:
info(f"{stock} is not in list of stocks") Parameters
----------
coins = set(re.findall(self.CRYPTO_REGEX, text)) text : str
for coin in coins: Blob of text.
if coin.lower() in self.crypto.symbol_list["symbol"].values:
symbols.append(Coin(coin.lower())) Returns
else: -------
info(f"{coin} is not in list of coins") list[Symbol]
List of stock symbols as Symbol objects
if symbols: """
info(symbols) schedule.run_pending()
return symbols symbols = []
stocks = set(re.findall(self.STOCK_REGEX, text))
def status(self, bot_resp) -> str: for stock in stocks:
"""Checks for any issues with APIs. if stock.upper() in self.stock.symbol_list["symbol"].values:
symbols.append(Stock(stock))
Returns else:
------- info(f"{stock} is not in list of stocks")
str
Human readable text on status of the bot and relevant APIs coins = set(re.findall(self.CRYPTO_REGEX, text))
""" for coin in coins:
if coin.lower() in self.crypto.symbol_list["symbol"].values:
stats = f""" symbols.append(Coin(coin.lower()))
Bot Status: else:
{bot_resp} info(f"{coin} is not in list of coins")
Stock Market Data: if symbols:
{self.stock.status()} info(symbols)
for symbol in symbols:
Cryptocurrency Data: self.trending_count[symbol.tag] = (
{self.crypto.status()} self.trending_count.get(symbol.tag, 0) + 1
""" )
warning(stats) return symbols
return stats def status(self, bot_resp) -> str:
"""Checks for any issues with APIs.
def search_symbols(self, search: str) -> list[tuple[str, str]]:
"""Performs a fuzzy search to find stock symbols closest to a search term. Returns
-------
Parameters str
---------- Human readable text on status of the bot and relevant APIs
search : str """
String used to search, could be a company name or something close to the companies stock ticker.
stats = f"""
Returns Bot Status:
------- {bot_resp}
list[tuple[str, str]]
A list tuples of every stock sorted in order of how well they match. Stock Market Data:
Each tuple contains: (Symbol, Issue Name). {self.stock.status()}
"""
Cryptocurrency Data:
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list]) {self.crypto.status()}
"""
search = search.lower()
warning(stats)
df["Match"] = df.apply(
lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()), return stats
axis=1,
) def search_symbols(self, search: str) -> list[tuple[str, str]]:
"""Performs a fuzzy search to find stock symbols closest to a search term.
df.sort_values(by="Match", ascending=False, inplace=True)
# if df["Match"].head().sum() < 300: Parameters
# df["Match"] = df.apply( ----------
# lambda x: fuzz.partial_ratio(search, x["name"].lower()), search : str
# axis=1, String used to search, could be a company name or something close to the companies stock ticker.
# )
Returns
# df.sort_values(by="Match", ascending=False, inplace=True) -------
list[tuple[str, str]]
symbols = df.head(20) A list tuples of every stock sorted in order of how well they match.
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"]))) Each tuple contains: (Symbol, Issue Name).
self.searched_symbols[search] = symbol_list """
return symbol_list
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
def inline_search(self, search: str) -> list[tuple[str, str]]:
"""Searches based on the shortest symbol that contains the same string as the search. search = search.lower()
Should be very fast compared to a fuzzy search.
df["Match"] = df.apply(
Parameters lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
---------- axis=1,
search : str )
String used to match against symbols.
df.sort_values(by="Match", ascending=False, inplace=True)
Returns # if df["Match"].head().sum() < 300:
------- # df["Match"] = df.apply(
list[tuple[str, str]] # lambda x: fuzz.partial_ratio(search, x["name"].lower()),
Each tuple contains: (Symbol, Issue Name). # axis=1,
""" # )
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list]) # df.sort_values(by="Match", ascending=False, inplace=True)
search = search.lower() symbols = df.head(20)
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
df = df[df["type_id"].str.contains(search, regex=False)].sort_values( self.searched_symbols[search] = symbol_list
by="type_id", key=lambda x: x.str.len() return symbol_list
)
def inline_search(self, search: str) -> list[tuple[str, str]]:
symbols = df.head(20) """Searches based on the shortest symbol that contains the same string as the search.
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"]))) Should be very fast compared to a fuzzy search.
self.searched_symbols[search] = symbol_list
return symbol_list Parameters
----------
def price_reply(self, symbols: list[Symbol]) -> list[str]: search : str
"""Returns current market price or after hours if its available for a given stock symbol. String used to match against symbols.
Parameters Returns
---------- -------
symbols : list list[tuple[str, str]]
List of stock symbols. Each tuple contains: (Symbol, Issue Name).
"""
Returns
------- df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
Dict[str, str]
Each symbol passed in is a key with its value being a human readable search = search.lower()
markdown formatted string of the symbols price and movement.
""" df = df[df["type_id"].str.contains(search, regex=False)].sort_values(
replies = [] by="type_id", key=lambda x: x.str.len()
)
for symbol in symbols:
info(symbol) symbols = df.head(20)
if isinstance(symbol, Stock): symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
replies.append(self.stock.price_reply(symbol)) self.searched_symbols[search] = symbol_list
elif isinstance(symbol, Coin): return symbol_list
replies.append(self.crypto.price_reply(symbol))
else: def price_reply(self, symbols: list[Symbol]) -> list[str]:
info(f"{symbol} is not a Stock or Coin") """Returns current market price or after hours if its available for a given stock symbol.
return replies Parameters
----------
def dividend_reply(self, symbols: list) -> list[str]: symbols : list
"""Returns the most recent, or next dividend date for a stock symbol. List of stock symbols.
Parameters Returns
---------- -------
symbols : list Dict[str, str]
List of stock symbols. Each symbol passed in is a key with its value being a human readable
markdown formatted string of the symbols price and movement.
Returns """
------- replies = []
Dict[str, str]
Each symbol passed in is a key with its value being a human readable for symbol in symbols:
formatted string of the symbols div dates. info(symbol)
""" if isinstance(symbol, Stock):
replies = [] replies.append(self.stock.price_reply(symbol))
for symbol in symbols: elif isinstance(symbol, Coin):
if isinstance(symbol, Stock): replies.append(self.crypto.price_reply(symbol))
replies.append(self.stock.dividend_reply(symbol)) else:
elif isinstance(symbol, Coin): info(f"{symbol} is not a Stock or Coin")
replies.append("Cryptocurrencies do no have Dividends.")
else: return replies
print(f"{symbol} is not a Stock or Coin")
def dividend_reply(self, symbols: list) -> list[str]:
return replies """Returns the most recent, or next dividend date for a stock symbol.
def news_reply(self, symbols: list) -> list[str]: Parameters
"""Gets recent english news on stock symbols. ----------
symbols : list
Parameters List of stock symbols.
----------
symbols : list Returns
List of stock symbols. -------
Dict[str, str]
Returns Each symbol passed in is a key with its value being a human readable
------- formatted string of the symbols div dates.
Dict[str, str] """
Each symbol passed in is a key with its value being a human replies = []
readable markdown formatted string of the symbols news. for symbol in symbols:
""" if isinstance(symbol, Stock):
replies = [] replies.append(self.stock.dividend_reply(symbol))
elif isinstance(symbol, Coin):
for symbol in symbols: replies.append("Cryptocurrencies do no have Dividends.")
if isinstance(symbol, Stock): else:
replies.append(self.stock.news_reply(symbol)) debug(f"{symbol} is not a Stock or Coin")
elif isinstance(symbol, Coin):
# replies.append(self.crypto.news_reply(symbol)) return replies
replies.append(
"News is not yet supported for cryptocurrencies. If you have any suggestions for news sources please contatct @MisterBiggs" def news_reply(self, symbols: list) -> list[str]:
) """Gets recent english news on stock symbols.
else:
print(f"{symbol} is not a Stock or Coin") Parameters
----------
return replies symbols : list
List of stock symbols.
def info_reply(self, symbols: list) -> list[str]:
"""Gets information on stock symbols. Returns
-------
Parameters Dict[str, str]
---------- Each symbol passed in is a key with its value being a human
symbols : list[str] readable markdown formatted string of the symbols news.
List of stock symbols. """
replies = []
Returns
------- for symbol in symbols:
Dict[str, str] if isinstance(symbol, Stock):
Each symbol passed in is a key with its value being a human readable formatted replies.append(self.stock.news_reply(symbol))
string of the symbols information. elif isinstance(symbol, Coin):
""" # replies.append(self.crypto.news_reply(symbol))
replies = [] replies.append(
"News is not yet supported for cryptocurrencies. If you have any suggestions for news sources please contatct @MisterBiggs"
for symbol in symbols: )
if isinstance(symbol, Stock): else:
replies.append(self.stock.info_reply(symbol)) debug(f"{symbol} is not a Stock or Coin")
elif isinstance(symbol, Coin):
replies.append(self.crypto.info_reply(symbol)) return replies
else:
print(f"{symbol} is not a Stock or Coin") def info_reply(self, symbols: list) -> list[str]:
"""Gets information on stock symbols.
return replies
Parameters
def intra_reply(self, symbol: Symbol) -> pd.DataFrame: ----------
"""Returns price data for a symbol since the last market open. symbols : list[str]
List of stock symbols.
Parameters
---------- Returns
symbol : str -------
Stock symbol. Dict[str, str]
Each symbol passed in is a key with its value being a human readable formatted
Returns string of the symbols information.
------- """
pd.DataFrame replies = []
Returns a timeseries dataframe with high, low, and volume data if its available.
Otherwise returns empty pd.DataFrame. for symbol in symbols:
""" if isinstance(symbol, Stock):
replies.append(self.stock.info_reply(symbol))
if isinstance(symbol, Stock): elif isinstance(symbol, Coin):
return self.stock.intra_reply(symbol) replies.append(self.crypto.info_reply(symbol))
elif isinstance(symbol, Coin): else:
return self.crypto.intra_reply(symbol) debug(f"{symbol} is not a Stock or Coin")
else:
print(f"{symbol} is not a Stock or Coin") return replies
return pd.DataFrame()
def intra_reply(self, symbol: Symbol) -> pd.DataFrame:
def chart_reply(self, symbol: Symbol) -> pd.DataFrame: """Returns price data for a symbol since the last market open.
"""Returns price data for a symbol of the past month up until the previous trading days close.
Also caches multiple requests made in the same day. Parameters
----------
Parameters symbol : str
---------- Stock symbol.
symbol : str
Stock symbol. Returns
-------
Returns pd.DataFrame
------- Returns a timeseries dataframe with high, low, and volume data if its available.
pd.DataFrame Otherwise returns empty pd.DataFrame.
Returns a timeseries dataframe with high, low, and volume data if its available. """
Otherwise returns empty pd.DataFrame.
""" if isinstance(symbol, Stock):
if isinstance(symbol, Stock): return self.stock.intra_reply(symbol)
return self.stock.chart_reply(symbol) elif isinstance(symbol, Coin):
elif isinstance(symbol, Coin): return self.crypto.intra_reply(symbol)
return self.crypto.chart_reply(symbol) else:
else: debug(f"{symbol} is not a Stock or Coin")
print(f"{symbol} is not a Stock or Coin") return pd.DataFrame()
return pd.DataFrame()
def chart_reply(self, symbol: Symbol) -> pd.DataFrame:
def stat_reply(self, symbols: list[Symbol]) -> list[str]: """Returns price data for a symbol of the past month up until the previous trading days close.
"""Gets key statistics for each symbol in the list Also caches multiple requests made in the same day.
Parameters Parameters
---------- ----------
symbols : list[str] symbol : str
List of stock symbols Stock symbol.
Returns Returns
------- -------
Dict[str, str] pd.DataFrame
Each symbol passed in is a key with its value being a human readable Returns a timeseries dataframe with high, low, and volume data if its available.
formatted string of the symbols statistics. Otherwise returns empty pd.DataFrame.
""" """
replies = [] if isinstance(symbol, Stock):
return self.stock.chart_reply(symbol)
for symbol in symbols: elif isinstance(symbol, Coin):
if isinstance(symbol, Stock): return self.crypto.chart_reply(symbol)
replies.append(self.stock.stat_reply(symbol)) else:
elif isinstance(symbol, Coin): debug(f"{symbol} is not a Stock or Coin")
replies.append(self.crypto.stat_reply(symbol)) return pd.DataFrame()
else:
print(f"{symbol} is not a Stock or Coin") def stat_reply(self, symbols: list[Symbol]) -> list[str]:
"""Gets key statistics for each symbol in the list
return replies
Parameters
def cap_reply(self, symbols: list[Symbol]) -> list[str]: ----------
"""Gets market cap for each symbol in the list symbols : list[str]
List of stock symbols
Parameters
---------- Returns
symbols : list[str] -------
List of stock symbols Dict[str, str]
Each symbol passed in is a key with its value being a human readable
Returns formatted string of the symbols statistics.
------- """
Dict[str, str] replies = []
Each symbol passed in is a key with its value being a human readable
formatted string of the symbols market cap. for symbol in symbols:
""" if isinstance(symbol, Stock):
replies = [] replies.append(self.stock.stat_reply(symbol))
elif isinstance(symbol, Coin):
for symbol in symbols: replies.append(self.crypto.stat_reply(symbol))
if isinstance(symbol, Stock): else:
replies.append(self.stock.cap_reply(symbol)) debug(f"{symbol} is not a Stock or Coin")
elif isinstance(symbol, Coin):
replies.append(self.crypto.cap_reply(symbol)) return replies
else:
print(f"{symbol} is not a Stock or Coin") def cap_reply(self, symbols: list[Symbol]) -> list[str]:
"""Gets market cap for each symbol in the list
return replies
Parameters
def trending(self) -> str: ----------
"""Checks APIs for trending symbols. symbols : list[str]
List of stock symbols
Returns
------- Returns
list[str] -------
List of preformatted strings to be sent to user. Dict[str, str]
""" Each symbol passed in is a key with its value being a human readable
formatted string of the symbols market cap.
stocks = self.stock.trending() """
coins = self.crypto.trending() replies = []
reply = "Trending Stocks:\n" for symbol in symbols:
reply += "-" * len("Trending Stocks:") + "\n" if isinstance(symbol, Stock):
for stock in stocks: replies.append(self.stock.cap_reply(symbol))
reply += stock + "\n" elif isinstance(symbol, Coin):
replies.append(self.crypto.cap_reply(symbol))
reply += "\n\nTrending Crypto:\n" else:
reply += "-" * len("Trending Crypto:") + "\n" debug(f"{symbol} is not a Stock or Coin")
for coin in coins:
reply += coin + "\n" return replies
return reply def spark_reply(self, symbols: list[Symbol]) -> list[str]:
"""Gets change for each symbol and returns it in a compact format
def random_pick(self) -> str:
Parameters
choice = random.choice( ----------
list(self.stock.symbol_list["description"]) symbols : list[str]
+ list(self.crypto.symbol_list["description"]) List of stock symbols
)
hold = ( Returns
datetime.date.today() + datetime.timedelta(random.randint(1, 365)) -------
).strftime("%b %d, %Y") list[str]
List of human readable strings.
return f"{choice}\nBuy and hold until: {hold}" """
replies = []
def batch_price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol. for symbol in symbols:
if isinstance(symbol, Stock):
Parameters replies.append(self.stock.spark_reply(symbol))
---------- elif isinstance(symbol, Coin):
symbols : list replies.append(self.crypto.spark_reply(symbol))
List of stock symbols. else:
debug(f"{symbol} is not a Stock or Coin")
Returns
------- return replies
Dict[str, str]
Each symbol passed in is a key with its value being a human readable def trending(self) -> str:
markdown formatted string of the symbols price and movement. """Checks APIs for trending symbols.
"""
replies = [] Returns
stocks = [] -------
coins = [] list[str]
List of preformatted strings to be sent to user.
for symbol in symbols: """
if isinstance(symbol, Stock):
stocks.append(symbol) stocks = self.stock.trending()
elif isinstance(symbol, Coin): coins = self.crypto.trending()
coins.append(symbol)
else: reply = ""
print(f"{symbol} is not a Stock or Coin")
if self.trending_count:
if stocks: reply += "🔥Trending on the Stock Bot:\n`"
# IEX batch endpoint doesnt seem to be working right now reply += "━" * len("Trending on the Stock Bot:") + "`\n"
for stock in stocks:
replies.append(self.stock.price_reply(stock)) sorted_trending = [
if coins: s[0]
replies = replies + self.crypto.batch_price(coins) for s in sorted(self.trending_count.items(), key=lambda item: item[1])
][::-1][0:5]
return replies
for t in sorted_trending:
reply += self.spark_reply(self.find_symbols(t))[0] + "\n"
if stocks:
reply += "\n\n💵Trending Stocks:\n`"
reply += "━" * len("Trending Stocks:") + "`\n"
for stock in stocks:
reply += stock + "\n"
if coins:
reply += "\n\n🦎Trending Crypto:\n`"
reply += "━" * len("Trending Crypto:") + "`\n"
for coin in coins:
reply += coin + "\n"
if "`$GME" in reply:
reply = reply.replace("🔥", "🦍")
if reply:
return reply
else:
warning("Failed to collect trending data.")
return "Trending data is not currently available."
def random_pick(self) -> str:
choice = random.choice(
list(self.stock.symbol_list["description"])
+ list(self.crypto.symbol_list["description"])
)
hold = (
datetime.date.today() + datetime.timedelta(random.randint(1, 365))
).strftime("%b %d, %Y")
return f"{choice}\nBuy and hold until: {hold}"
def batch_price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol.
Parameters
----------
symbols : list
List of stock symbols.
Returns
-------
Dict[str, str]
Each symbol passed in is a key with its value being a human readable
markdown formatted string of the symbols price and movement.
"""
replies = []
stocks = []
coins = []
for symbol in symbols:
if isinstance(symbol, Stock):
stocks.append(symbol)
elif isinstance(symbol, Coin):
coins.append(symbol)
else:
debug(f"{symbol} is not a Stock or Coin")
if stocks:
# IEX batch endpoint doesnt seem to be working right now
for stock in stocks:
replies.append(self.stock.price_reply(stock))
if coins:
replies = replies + self.crypto.batch_price(coins)
return replies