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 15:06:53 +00:00
Code Releases Activity

I think i wrangled the types

Browse Source
This commit is contained in:
Anson Biggs 2021-02-28 13:54:46 -07:00
parent 19c2daa9f9
commit f7dc433ec7
3 changed files with 68 additions and 116 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
Symbol.py
View File

@ -1,27 +1,49 @@
import requests as r
from cg_Crypto import cg_Crypto
class Symbol:
"""
symbol: What the user calls it. ie tsla or btc
id: What the api expects. ie tsla or bitcoin
name: Human readable. ie Tesla or Bitcoin
"""
currency = "usd"
pass
class Stock(Symbol):
def __init__(self, symbol) -> None:
self.symbol = symbol
self.id = symbol
self.name = symbol
def __repr__(self) -> str:
return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>"
def __str__(self) -> str:
return self.id
class Stock(Symbol):
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.id = symbol
# This is so every Coin instance doesnt have to download entire list of coin symbols and id's
cg = cg_Crypto()
class Coin(Symbol):
def __init__(self, symbol) -> None:
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.get_data()
def get_data(self) -> None:
data = r.get("https://api.coingecko.com/api/v3/coins/" + self.symbol).json()
self.id = cg.symbol_id(self.symbol)
data = r.get("https://api.coingecko.com/api/v3/coins/" + self.id).json()
self.data = data
self.id = data["id"]
self.name = data["name"]
self.description = data["description"]
self.price = data["market_data"]["current_price"][self.currency]
self.data = data

82
bot.py
View File

@ -3,7 +3,6 @@ import datetime
import io
import logging
import os
import random
import html
import json
import traceback
@ -56,18 +55,15 @@ def start(update: Update, context: CallbackContext):
def help(update: Update, context: CallbackContext):
"""Send link to docs when the command /help is issued."""
update.message.reply_text(text=t.help_text, parse_mode=telegram.ParseMode.MARKDOWN)
def license(update: Update, context: CallbackContext):
"""Return bots license agreement"""
update.message.reply_text(text=t.license, parse_mode=telegram.ParseMode.MARKDOWN)
def status(update: Update, context: CallbackContext):
update.message.reply_text(text=s.status(), parse_mode=telegram.ParseMode.MARKDOWN)
@ -139,7 +135,6 @@ def symbol_detect(update: Update, context: CallbackContext):
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
print(symbols)
for reply in s.price_reply(symbols):
print(reply)
update.message.reply_text(
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
)
@ -162,9 +157,9 @@ def dividend(update: Update, context: CallbackContext):
if symbols:
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for symbol in symbols:
for reply in s.dividend_reply(symbols):
update.message.reply_text(
text=s.dividend_reply(symbol), parse_mode=telegram.ParseMode.MARKDOWN
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
)
@ -186,9 +181,9 @@ def news(update: Update, context: CallbackContext):
if symbols:
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in s.news_reply(symbols).items():
for reply in s.news_reply(symbols):
update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
)
@ -210,9 +205,9 @@ def info(update: Update, context: CallbackContext):
if symbols:
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in s.info_reply(symbols).items():
for reply in s.info_reply(symbols):
update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
)
@ -247,8 +242,8 @@ def intra(update: Update, context: CallbackContext):
)
return
symbol = s.find_symbols(message)
symbols = s.find_symbols(message)
symbol = symbols[0]
df = s.intra_reply(symbol)
if df.empty:
update.message.reply_text(
@ -265,7 +260,7 @@ def intra(update: Update, context: CallbackContext):
mpf.plot(
df,
type="renko",
title=f"\n${symbol.upper()}",
title=f"\n${symbol.name}",
volume="volume" in df.keys(),
style="yahoo",
mav=20,
@ -275,9 +270,9 @@ def intra(update: Update, context: CallbackContext):
update.message.reply_photo(
photo=buf,
caption=f"\nIntraday chart for ${symbol.upper()} from {df.first_valid_index().strftime('%I:%M')} to"
caption=f"\nIntraday chart for ${symbol.name} from {df.first_valid_index().strftime('%I:%M')} to"
+ f" {df.last_valid_index().strftime('%I:%M')} ET on"
+ f" {datetime.date.today().strftime('%d, %b %Y')}\n\n{s.price_reply([symbol])[symbol]}",
+ f" {datetime.date.today().strftime('%d, %b %Y')}\n\n{s.price_reply([symbol])[0]}",
parse_mode=telegram.ParseMode.MARKDOWN,
)
@ -295,8 +290,8 @@ def chart(update: Update, context: CallbackContext):
return
symbols = s.find_symbols(message)
df, symbol = s.chart_reply(symbols)
symbol = symbols[0]
df = s.chart_reply(symbol)
if df.empty:
update.message.reply_text(
text="Invalid symbol please see `/help` for usage details.",
@ -312,7 +307,7 @@ def chart(update: Update, context: CallbackContext):
mpf.plot(
df,
type="candle",
title=f"\n${symbol.upper()}",
title=f"\n${symbol.name}",
volume="volume" in df.keys(),
style="yahoo",
savefig=dict(fname=buf, dpi=400, bbox_inches="tight"),
@ -321,8 +316,8 @@ def chart(update: Update, context: CallbackContext):
update.message.reply_photo(
photo=buf,
caption=f"\n1 Month chart for ${symbol.upper()} from {df.first_valid_index().strftime('%d, %b %Y')}"
+ f" to {df.last_valid_index().strftime('%d, %b %Y')}\n\n{s.price_reply(symbols)[0]}",
caption=f"\n1 Month chart for ${symbol.name} from {df.first_valid_index().strftime('%d, %b %Y')}"
+ f" to {df.last_valid_index().strftime('%d, %b %Y')}\n\n{s.price_reply([symbol])[0]}",
parse_mode=telegram.ParseMode.MARKDOWN,
)
@ -345,38 +340,12 @@ def stat(update: Update, context: CallbackContext):
if symbols:
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
for reply in s.stat_reply(symbols).items():
for reply in s.stat_reply(symbols):
update.message.reply_text(
text=reply[1], parse_mode=telegram.ParseMode.MARKDOWN
text=reply, parse_mode=telegram.ParseMode.MARKDOWN
)
def crypto(update: Update, context: CallbackContext):
"""
https://iexcloud.io/docs/api/#cryptocurrency-quote
"""
context.bot.send_chat_action(
chat_id=update.message.chat_id, action=telegram.ChatAction.TYPING
)
message = update.message.text
if message.strip() == "/crypto":
update.message.reply_text(
"This command returns the current price in USD for a cryptocurrency.\nExample: /crypto eth"
)
return
reply = s.crypto_reply(message)
if reply:
update.message.reply_text(text=reply, parse_mode=telegram.ParseMode.MARKDOWN)
else:
update.message.reply_text(
text=f"Pair: {message} returned an error.",
parse_mode=telegram.ParseMode.MARKDOWN,
)
def inline_query(update: Update, context: CallbackContext):
"""
Handles inline query.
@ -388,7 +357,7 @@ def inline_query(update: Update, context: CallbackContext):
results = []
for match in matches:
try:
price = s.price_reply([match[0]])[match[0]]
price = s.price_reply([match[0]])[0]
results.append(
InlineQueryResultArticle(
match[0],
@ -409,13 +378,8 @@ def inline_query(update: Update, context: CallbackContext):
def rand_pick(update: Update, context: CallbackContext):
choice = random.choice(list(s.symbol_list["description"]))
hold = (
datetime.date.today() + datetime.timedelta(random.randint(1, 365))
).strftime("%b %d, %Y")
update.message.reply_text(
text=f"{choice}\nBuy and hold until: {hold}",
text=s.random_pick(),
parse_mode=telegram.ParseMode.MARKDOWN,
)
@ -467,8 +431,7 @@ def main():
dp.add_handler(CommandHandler("search", search))
dp.add_handler(CommandHandler("intraday", intra))
dp.add_handler(CommandHandler("intra", intra, run_async=True))
dp.add_handler(CommandHandler("chart", chart))
dp.add_handler(CommandHandler("crypto", crypto))
dp.add_handler(CommandHandler("chart", chart, run_async=True))
dp.add_handler(CommandHandler("random", rand_pick))
dp.add_handler(CommandHandler("donate", donate))
dp.add_handler(CommandHandler("status", status))
@ -493,9 +456,6 @@ def main():
# Start the Bot
updater.start_polling()
# Run the bot until you press Ctrl-C or the process receives SIGINT,
# SIGTERM or SIGABRT. This should be used most of the time, since
# start_polling() is non-blocking and will stop the bot gracefully.
updater.idle()

66
symbol_router.py
View File

@ -2,14 +2,17 @@
"""
import re
import requests as r
import pandas as pd
import random
import datetime
from typing import List, Tuple, TypeVar
from typing import List, Tuple
from IEX_Symbol import IEX_Symbol
from cg_Crypto import cg_Crypto
from Symbol import Symbol, Stock, Coin
class Router:
STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z]{1,4})"
@ -19,7 +22,7 @@ class Router:
self.stock = IEX_Symbol()
self.crypto = cg_Crypto()
def find_symbols(self, text: str) -> List[any]:
def find_symbols(self, text: str) -> List[Symbol]:
"""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.
Only returns each match once. Example: Whats the price of $tsla?
@ -177,7 +180,7 @@ class Router:
return replies
def intra_reply(self, symbol: str) -> pd.DataFrame:
def intra_reply(self, symbol: Symbol) -> pd.DataFrame:
"""Returns price data for a symbol since the last market open.
Parameters
@ -199,7 +202,7 @@ class Router:
print(f"{symbol} is not a Stock or Coin")
return pd.DataFrame()
def chart_reply(self, symbol: str) -> pd.DataFrame:
def chart_reply(self, symbol: Symbol) -> 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.
@ -221,7 +224,7 @@ class Router:
print(f"{symbol} is not a Stock or Coin")
return pd.DataFrame()
def stat_reply(self, symbols: List[str]) -> List[str]:
def stat_reply(self, symbols: List[Symbol]) -> List[str]:
"""Gets key statistics for each symbol in the list
Parameters
@ -246,47 +249,14 @@ class Router:
return replies
def random_pick(self) -> str:
Sym = TypeVar("Sym", Stock, Coin)
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")
class Symbol:
"""
symbol: What the user calls it. ie tsla or btc
id: What the api expects. ie tsla or bitcoin
name: Human readable. ie Tesla or Bitcoin
"""
currency = "usd"
pass
def __init__(self, symbol) -> None:
self.symbol = symbol
self.id = symbol
def __repr__(self) -> str:
return f"<{self.__class__.__name__} instance of {self.id} at {id(self)}>"
def __str__(self) -> str:
return self.id
class Stock(Symbol):
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.id = symbol
class Coin(Symbol):
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.get_data()
def get_data(self) -> None:
self.id = cg_Crypto().symbol_id(self.symbol)
data = r.get("https://api.coingecko.com/api/v3/coins/" + self.id).json()
self.data = data
self.name = data["name"]
self.description = data["description"]
self.price = data["market_data"]["current_price"][self.currency]
return f"{choice}\nBuy and hold until: {hold}"