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

Merge branch 'canary' into 'master'

Browse Source 
Change to API requests and small formatting changes

See merge request simple-stock-bots/simple-telegram-stock-bot!26
This commit is contained in:
Anson Biggs 2021-08-16 02:35:39 +00:00
commit a09f8aec24
7 changed files with 214 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
.devcontainer/devcontainer.json Normal file
View File

@ -0,0 +1,25 @@
// For format details, see https://aka.ms/devcontainer.json. For config options, see the README at:
// https://github.com/microsoft/vscode-dev-containers/tree/v0.191.0/containers/docker-existing-dockerfile
{
"name": "DockerDev",
// Sets the run context to one level up instead of the .devcontainer folder.
"context": "..",
// Update the 'dockerFile' property if you aren't using the standard 'Dockerfile' filename.
"dockerFile": "../DockerDev",
// Set *default* container specific settings.json values on container create.
"settings": {},
// Add the IDs of extensions you want installed when the container is created.
"extensions": [
"ms-python.python"
]
// Use 'forwardPorts' to make a list of ports inside the container available locally.
// "forwardPorts": [],
// Uncomment the next line to run commands after the container is created - for example installing curl.
// "postCreateCommand": "apt-get update && apt-get install -y curl",
// Uncomment when using a ptrace-based debugger like C++, Go, and Rust
// "runArgs": [ "--cap-add=SYS_PTRACE", "--security-opt", "seccomp=unconfined" ],
// Uncomment to use the Docker CLI from inside the container. See https://aka.ms/vscode-remote/samples/docker-from-docker.
// "mounts": [ "source=/var/run/docker.sock,target=/var/run/docker.sock,type=bind" ],
// Uncomment to connect as a non-root user if you've added one. See https://aka.ms/vscode-remote/containers/non-root.
// "remoteUser": "vscode"
}

3
.gitignore vendored
View File

@ -1,2 +1 @@
__pycache__/
.vscode
__pycache__

20
.vscode/settings.json vendored Normal file
View File

@ -0,0 +1,20 @@
{
"workbench.iconTheme": "vscode-icons",
"editor.suggestSelection": "first",
"vsintellicode.modify.editor.suggestSelection": "automaticallyOverrodeDefaultValue",
"python.languageServer": "Pylance",
"git.autofetch": true,
"editor.formatOnSave": true,
"files.associations": {
"DockerDev": "dockerfile",
},
"workbench.colorTheme": "One Dark Pro",
"explorer.confirmDelete": false,
"editor.fontFamily": "JetBrains Mono",
"editor.letterSpacing": 1.2,
"editor.fontLigatures": true,
"editor.wordWrap": "on",
"python.formatting.provider": "black",
"python.showStartPage": false,
"editor.fontSize": 13,
}

7
DockerDev Normal file
View File

@ -0,0 +1,7 @@
FROM python:3.9-buster
COPY requirements.txt ./
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir black
COPY . .

162
IEX_Symbol.py
View File

@ -1,6 +1,7 @@
"""Class with functions for running the bot with IEX Cloud.
"""
import logging
import os
from datetime import datetime
from logging import warning
@ -47,6 +48,30 @@ class IEX_Symbol:
schedule.every().day.do(self.get_symbol_list)
schedule.every().day.do(self.clear_charts)
def get(self, endpoint, params: dict = {}, timeout=5) -> dict:
url = "https://cloud.iexapis.com/stable" + endpoint
# set token param if it wasn't passed.
params["token"] = params.get("token", self.IEX_TOKEN)
resp = r.get(url, params=params, timeout=timeout)
# Make sure API returned a proper status code
try:
resp.raise_for_status()
except r.exceptions.HTTPError as e:
logging.error(e)
return {}
# Make sure API returned valid JSON
try:
resp_json = resp.json()
return resp_json
except r.exceptions.JSONDecodeError as e:
logging.error(e)
return {}
def clear_charts(self) -> None:
"""
Clears cache of chart data.
@ -71,14 +96,8 @@ class IEX_Symbol:
If `return_df` is set to `True` returns a dataframe, otherwise returns `None`.
"""
reg_symbols = r.get(
f"https://cloud.iexapis.com/stable/ref-data/symbols?token={self.IEX_TOKEN}",
timeout=5,
).json()
otc_symbols = r.get(
f"https://cloud.iexapis.com/stable/ref-data/otc/symbols?token={self.IEX_TOKEN}",
timeout=5,
).json()
reg_symbols = self.get("/ref-data/symbols")
otc_symbols = self.get("/ref-data/otc/symbols")
reg = pd.DataFrame(data=reg_symbols)
otc = pd.DataFrame(data=otc_symbols)
@ -105,7 +124,7 @@ class IEX_Symbol:
"""
resp = r.get(
"https://pjmps0c34hp7.statuspage.io/api/v2/status.json",
timeout=5,
timeout=15,
)
if resp.status_code == 200:
@ -174,14 +193,7 @@ class IEX_Symbol:
Formatted markdown
"""
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol.id}/quote?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200:
IEXData = response.json()
if IEXData := self.get(f"/stock/{symbol.id}/quote"):
if symbol.symbol.upper() in self.otc_list:
return f"OTC - {symbol.symbol.upper()}, {IEXData['companyName']} most recent price is: $**{IEXData['latestPrice']}**"
@ -196,19 +208,20 @@ class IEX_Symbol:
if set(keys).issubset(IEXData):
if change := IEXData.get("changePercent", 0):
change = round(change * 100, 2)
else:
change = 0
if (
IEXData.get("isUSMarketOpen", True)
or (IEXData["extendedChangePercent"] is None)
or (IEXData["extendedPrice"] is None)
): # Check if market is open.
message = f"The current stock price of {IEXData['companyName']} is $**{IEXData['latestPrice']}**"
if change := IEXData.get("changePercent", 0):
change = round(change * 100, 2)
else:
change = 0
else:
message = (
f"{IEXData['companyName']} closed at $**{IEXData['latestPrice']}**,"
f"{IEXData['companyName']} closed at $**{IEXData['latestPrice']}** with a change of {change}%,"
+ f" after hours _(15 minutes delayed)_ the stock price is $**{IEXData['extendedPrice']}**"
)
if change := IEXData.get("extendedChangePercent", 0):
@ -248,13 +261,11 @@ class IEX_Symbol:
if symbol.symbol.upper() in self.otc_list:
return "OTC stocks do not currently support any commands."
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/dividends/next?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200 and response.json():
IEXData = response.json()[0]
if resp := self.get(f"/stock/{symbol.id}/dividends/next"):
try:
IEXData = resp[0]
except IndexError as e:
return f"${symbol.id.upper()} either doesn't exist or pays no dividend."
keys = (
"amount",
"currency",
@ -312,28 +323,19 @@ class IEX_Symbol:
if symbol.symbol.upper() in self.otc_list:
return "OTC stocks do not currently support any commands."
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/news/last/15?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200:
data = response.json()
if data:
line = []
if data := self.get(f"/stock/{symbol.id}/news/last/15"):
line = []
for news in data:
if news["lang"] == "en" and not news["hasPaywall"]:
line.append(
f"*{news['source']}*: [{news['headline']}]({news['url']})"
)
for news in data:
if news["lang"] == "en" and not news["hasPaywall"]:
line.append(
f"*{news['source']}*: [{news['headline']}]({news['url']})"
)
return f"News for **{symbol.id.upper()}**:\n" + "\n".join(line[:5])
else:
return f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
else:
return f"No news found for: {symbol}\nEither today is boring or the symbol does not exist."
return f"News for **{symbol.id.upper()}**:\n" + "\n".join(line[:5])
return f"No news found for: {symbol.id}\nEither today is boring or the symbol does not exist."
def info_reply(self, symbol: Stock) -> str:
"""Gets description for Stock
@ -350,14 +352,7 @@ class IEX_Symbol:
if symbol.symbol.upper() in self.otc_list:
return "OTC stocks do not currently support any commands."
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/company?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200:
data = response.json()
if data := self.get(f"/stock/{symbol.id}/company"):
[data.pop(k) for k in list(data) if data[k] == ""]
if "description" in data:
@ -380,14 +375,7 @@ class IEX_Symbol:
if symbol.symbol.upper() in self.otc_list:
return "OTC stocks do not currently support any commands."
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/stats?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200:
data = response.json()
if data := self.get(f"/stock/{symbol.id}/stats"):
[data.pop(k) for k in list(data) if data[k] == ""]
m = ""
@ -411,25 +399,20 @@ class IEX_Symbol:
else:
return f"No information found for: {symbol}\nEither today is boring or the symbol does not exist."
def cap_reply(self, stock: Stock) -> str:
def cap_reply(self, symbol: Stock) -> str:
"""Get the Market Cap of a stock"""
response = r.get(
f"https://cloud.iexapis.com/stable/stock/{stock.id}/stats?token={self.IEX_TOKEN}",
timeout=5,
)
if response.status_code == 200:
if data := self.get(f"/stable/stock/{symbol.id}/stats"):
try:
data = response.json()
cap = data["marketcap"]
except KeyError:
return f"{stock.id} returned an error."
return f"{symbol.id} returned an error."
message = f"The current market cap of {stock.name} is $**{cap:,.2f}**"
message = f"The current market cap of {symbol.name} is $**{cap:,.2f}**"
else:
message = f"The Coin: {stock.name} was not found or returned and error."
message = f"The Stock: {symbol.name} was not found or returned and error."
return message
@ -452,13 +435,8 @@ class IEX_Symbol:
if symbol.id.upper() not in list(self.symbol_list["symbol"]):
return pd.DataFrame()
IEXurl = f"https://cloud.iexapis.com/stable/stock/{symbol}/intraday-prices?token={self.IEX_TOKEN}"
response = r.get(
IEXurl,
timeout=5,
)
if response.status_code == 200:
df = pd.DataFrame(response.json())
if data := self.get(f"/stock/{symbol.id}/intraday-prices"):
df = pd.DataFrame(data)
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
df = df.set_index("DT")
@ -493,13 +471,11 @@ class IEX_Symbol:
except KeyError:
pass
response = r.get(
f"https://cloud.iexapis.com/stable/stock/{symbol}/chart/1mm?token={self.IEX_TOKEN}&chartInterval=3&includeToday=false",
timeout=5,
)
if response.status_code == 200:
df = pd.DataFrame(response.json())
if data := self.get(
f"/stock/{symbol.id}/chart/1mm",
params={"chartInterval": 3, "includeToday": "false"},
):
df = pd.DataFrame(data)
df.dropna(inplace=True, subset=["date", "minute", "high", "low", "volume"])
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
df = df.set_index("DT")
@ -517,14 +493,10 @@ class IEX_Symbol:
list of $ID: NAME, CHANGE%
"""
stocks = r.get(
f"https://cloud.iexapis.com/stable/stock/market/list/mostactive?token={self.IEX_TOKEN}",
timeout=5,
)
if stocks.status_code == 200:
if data := self.get(f"/stock/market/list/mostactive"):
return [
f"`${s['symbol']}`: {s['companyName']}, {100*s['changePercent']:.2f}%"
for s in stocks.json()
for s in data
]
else:
return ["Trending Stocks Currently Unavailable."]

2
bot.py
View File

@ -101,7 +101,7 @@ def donate(update: Update, context: CallbackContext):
info(f"Donate command ran by {update.message.chat.username}")
chat_id = update.message.chat_id
if update.message.text.strip() == "/donate":
if update.message.text.strip() == "/donate" or "/donate@" in update.message.text:
update.message.reply_text(
text=t.donate_text,
parse_mode=telegram.ParseMode.MARKDOWN,

160
cg_Crypto.py
View File

@ -1,6 +1,7 @@
"""Class with functions for running the bot with IEX Cloud.
"""
import logging
from datetime import datetime
from typing import List, Optional, Tuple
@ -33,6 +34,25 @@ class cg_Crypto:
self.get_symbol_list()
schedule.every().day.do(self.get_symbol_list)
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)
# Make sure API returned a proper status code
try:
resp.raise_for_status()
except r.exceptions.HTTPError as e:
logging.error(e)
return {}
# Make sure API returned valid JSON
try:
resp_json = resp.json()
return resp_json
except r.exceptions.JSONDecodeError as e:
logging.error(e)
return {}
def symbol_id(self, symbol) -> str:
try:
return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[
@ -45,10 +65,7 @@ class cg_Crypto:
self, return_df=False
) -> Optional[Tuple[pd.DataFrame, datetime]]:
raw_symbols = r.get(
"https://api.coingecko.com/api/v3/coins/list",
timeout=5,
).json()
raw_symbols = self.get("/coins/list")
symbols = pd.DataFrame(data=raw_symbols)
symbols["description"] = (
@ -69,15 +86,16 @@ class cg_Crypto:
str
Human readable text on status of CoinGecko API
"""
status = r.get(
"https://api.coingecko.com/api/v3/ping",
status = self.get(
"/ping",
timeout=5,
)
if status.status_code == 200:
return f"CoinGecko API responded that it was OK in {status.elapsed.total_seconds()} Seconds."
else:
return f"CoinGecko API returned an error in {status.elapsed.total_seconds()} Seconds."
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."
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.
@ -133,14 +151,16 @@ class cg_Crypto:
markdown formatted string of the symbols price and movement.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/simple/price?ids={coin.id}&vs_currencies={self.vs_currency}&include_24hr_change=true",
timeout=5,
)
if response.status_code == 200:
if resp := self.get(
"/simple/price",
params={
"ids": coin.id,
"vs_currencies": self.vs_currency,
"include_24hr_change": "true",
},
):
try:
data = response.json()[coin.id]
data = resp[coin.id]
price = data[self.vs_currency]
change = data[self.vs_currency + "_24h_change"]
@ -160,7 +180,7 @@ class cg_Crypto:
message += ", the coin hasn't shown any movement today."
else:
message = f"The Coin: {coin.name} was not found."
message = f"The price for {coin.name} is not available. If you suspect this is an error run `/status`"
return message
@ -177,13 +197,13 @@ class cg_Crypto:
pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol.id}/ohlc?vs_currency=usd&days=1",
timeout=5,
)
if response.status_code == 200:
if resp := self.get(
f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 1},
):
df = pd.DataFrame(
response.json(), columns=["Date", "Open", "High", "Low", "Close"]
resp, columns=["Date", "Open", "High", "Low", "Close"]
).dropna()
df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date")
@ -205,14 +225,13 @@ class cg_Crypto:
pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol.id}/ohlc?vs_currency=usd&days=30",
timeout=5,
)
if response.status_code == 200:
if resp := self.get(
f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 30},
):
df = pd.DataFrame(
response.json(), columns=["Date", "Open", "High", "Low", "Close"]
resp, columns=["Date", "Open", "High", "Low", "Close"]
).dropna()
df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date")
@ -233,12 +252,12 @@ class cg_Crypto:
Preformatted markdown.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol.id}?localization=false",
timeout=5,
)
if response.status_code == 200:
data = response.json()
if data := self.get(
f"/coins/{symbol.id}",
params={
"localization": "false",
},
):
return f"""
[{data['name']}]({data['links']['homepage'][0]}) Statistics:
@ -265,14 +284,18 @@ class cg_Crypto:
str
Preformatted markdown.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/simple/price?ids={coin.id}&vs_currencies={self.vs_currency}&include_market_cap=true",
timeout=5,
)
if response.status_code == 200:
if resp := self.get(
f"/simple/price",
params={
"ids": coin.id,
"vs_currencies": self.vs_currency,
"include_market_cap": "true",
},
):
print(resp)
try:
data = response.json()[coin.id]
data = resp[coin.id]
price = data[self.vs_currency]
cap = data[self.vs_currency + "_market_cap"]
@ -302,12 +325,10 @@ class cg_Crypto:
Preformatted markdown.
"""
response = r.get(
f"https://api.coingecko.com/api/v3/coins/{symbol.id}?localization=false",
timeout=5,
)
if response.status_code == 200:
data = response.json()
if data := self.get(
f"/coins/{symbol.id}",
params={"localization": "false"},
):
try:
return markdownify(data["description"]["en"])
except KeyError:
@ -324,28 +345,29 @@ class cg_Crypto:
list of $$ID: NAME, CHANGE%
"""
coins = r.get(
"https://api.coingecko.com/api/v3/search/trending",
timeout=5,
)
coins = self.get("/search/trending")
try:
trending = []
if coins.status_code == 200:
for coin in coins.json()["coins"]:
c = coin["item"]
for coin in coins["coins"]:
c = coin["item"]
sym = c["symbol"].upper()
name = c["name"]
change = r.get(
f"https://api.coingecko.com/api/v3/simple/price?ids={c['id']}&vs_currencies={self.vs_currency}&include_24hr_change=true"
).json()[c["id"]]["usd_24h_change"]
sym = c["symbol"].upper()
name = c["name"]
change = self.get(
f"/simple/price",
params={
"ids": c["id"],
"vs_currencies": self.vs_currency,
"include_24hr_change": "true",
},
)[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:
print(e)
logging.warning(e)
trending = ["Trending Coins Currently Unavailable."]
return trending
@ -364,10 +386,14 @@ class cg_Crypto:
"""
query = ",".join([c.id for c in coins])
prices = r.get(
f"https://api.coingecko.com/api/v3/simple/price?ids={query}&vs_currencies=usd&include_24hr_change=true",
timeout=5,
).json()
prices = self.get(
f"/simple/price",
params={
"ids": query,
"vs_currencies": self.vs_currency,
"include_24hr_change": "true",
},
)
replies = []
for coin in coins: