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

#18

Browse Source
This commit is contained in:
Anson 2021-09-01 01:51:18 +00:00
parent 0877949a89
commit dcce2cb95a
1 changed files with 413 additions and 412 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

825
cg_Crypto.py
View File

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