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

Merge branch 'Canary' into 'master'

Browse Source 
Canary

Closes #76 and #78

See merge request simple-stock-bots/simple-telegram-stock-bot!34
This commit is contained in:
Anson Biggs 2021-11-06 20:22:00 +00:00
commit 579421f4ab
10 changed files with 589 additions and 641 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.devcontainer/Dockerfile
View File

@ -7,9 +7,13 @@ RUN pip install --user -r requirements.txt
FROM python:3.9-slim
COPY --from=builder /root/.local /root/.local
RUN pip install --no-cache-dir black
ENV MPLBACKEND=Agg
COPY --from=builder /root/.local /root/.local
RUN pip install --no-cache-dir black
ENV TELEGRAM=TOKEN
ENV IEX=TOKEN
COPY . .

3
.devcontainer/devcontainer.json
View File

@ -10,7 +10,8 @@
"settings": {},
// Add the IDs of extensions you want installed when the container is created.
"extensions": [
"ms-python.python"
"ms-python.python",
"ms-azuretools.vscode-docker"
]
// Use 'forwardPorts' to make a list of ports inside the container available locally.
// "forwardPorts": [],

15
.vscode/launch.json vendored Normal file
View File

@ -0,0 +1,15 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"name": "Telegram Bot",
"type": "python",
"request": "launch",
"program": "bot.py",
"console": "integratedTerminal"
}
]
}

2
Dockerfile
View File

@ -7,6 +7,8 @@ RUN pip install --user -r requirements.txt
FROM python:3.9-slim
ENV MPLBACKEND=Agg
COPY --from=builder /root/.local /root/.local

1007
IEX_Symbol.py
View File

File diff suppressed because it is too large Load Diff

23
T_info.py
View File

@ -39,7 +39,7 @@ Simply calling a symbol in any message that the bot can see will also return the
- `/help` Get some help using the bot. 🆘
**Inline Features**
You can type @SimpleStockBot `[search]` in any chat or direct message to search for the stock bots full list of stock symbols and return the price of the ticker. Then once you select the ticker want the bot will send a message as you in that chat with the latest stock price.
You can type @SimpleStockBot `[search]` in any chat or direct message to search for the stock bots full list of stock and crypto symbols and return the price. Then once you select the ticker want the bot will send a message as you in that chat with the latest stock price. Prices may be delayed by up to an hour.
Market data is provided by [IEX Cloud](https://iexcloud.io)
@ -73,3 +73,24 @@ trending - Trending Stocks and Cryptos. 💬
intra - $[symbol] Plot since the last market open. 📈
chart - $[chart] Plot of the past month. 📊
""" # Not used by the bot but for updaing commands with BotFather
tests = """
/info $tsla
/info $$btc
/news $tsla
/news $$btc
/stat $tsla
/stat $$btc
/cap $tsla
/cap $$btc
/dividend $tsla
/dividend $msft
/dividend $$btc
/intra $tsla
/intra $$btc
/chart $tsla
/chart $$btc
/help
/trending
"""

72
bot.py
View File

@ -85,14 +85,17 @@ def license(update: Update, context: CallbackContext):
def status(update: Update, context: CallbackContext):
"""Gather status of bot and dependant services and return important status updates."""
warning(f"Status command ran by {update.message.chat.username}")
bot_resp = datetime.datetime.now(update.message.date.tzinfo) - update.message.date
bot_resp_time = (
datetime.datetime.now(update.message.date.tzinfo) - update.message.date
)
bot_status = s.status(
f"It took {bot_resp_time.total_seconds()} seconds for the bot to get your message."
)
update.message.reply_text(
text=s.status(
f"It took {bot_resp.total_seconds()} seconds for the bot to get your message."
),
text=bot_status,
parse_mode=telegram.ParseMode.MARKDOWN,
disable_notification=True,
)
@ -261,8 +264,8 @@ def information(update: Update, context: CallbackContext):
def search(update: Update, context: CallbackContext):
"""
Uses fuzzy search on full list of stocks and crypto names
and descriptions then returns the top matches in order.
Searches on full list of stocks and crypto descriptions
then returns the top matches in order of smallest symbol name length.
"""
info(f"Search command ran by {update.message.chat.username}")
message = update.message.text.replace("/search ", "")
@ -275,11 +278,13 @@ def search(update: Update, context: CallbackContext):
return
context.bot.send_chat_action(chat_id=chat_id, action=telegram.ChatAction.TYPING)
queries = s.search_symbols(message)[:10]
if queries:
reply = "*Search Results:*\n`$ticker: Company Name`\n`" + ("-" * 21) + "`\n"
for query in queries:
reply += "`" + query[1] + "`\n"
queries = s.inline_search(message, matches=10)
if not queries.empty:
reply = "*Search Results:*\n`$ticker` : Company Name\n`" + ("-" * 21) + "`\n"
for _, query in queries.iterrows():
desc = query["description"]
reply += "`" + desc.replace(": ", "` : ") + "\n"
update.message.reply_text(
text=reply,
parse_mode=telegram.ParseMode.MARKDOWN,
@ -466,31 +471,45 @@ def inline_query(update: Update, context: CallbackContext):
Handles inline query. Searches by looking if query is contained
in the symbol and returns matches in alphabetical order.
"""
info(f"Inline command ran by {update.message.chat.username}")
# info(f"Inline command ran by {update.message.chat.username}")
info(f"Query: {update.inline_query.query}")
matches = s.inline_search(update.inline_query.query)[:5]
symbols = " ".join([match[1].split(":")[0] for match in matches])
prices = s.batch_price_reply(s.find_symbols(symbols))
ignored_queries = {"$", "$$", " ", ""}
results = []
for match, price in zip(matches, prices):
try:
results.append(
if update.inline_query.query.strip() in ignored_queries:
default_message = """
You can type:\n@SimpleStockBot `[search]`\nin any chat or direct message to search for the stock bots full list of stock and crypto symbols and return the price.
"""
update.inline_query.answer(
[
InlineQueryResultArticle(
str(uuid4()),
title=match[1],
title="Please enter a query. It can be a ticker or a name of a company.",
input_message_content=InputTextMessageContent(
price, parse_mode=telegram.ParseMode.MARKDOWN
default_message, parse_mode=telegram.ParseMode.MARKDOWN
),
)
]
)
matches = s.inline_search(update.inline_query.query)
results = []
for _, row in matches.iterrows():
results.append(
InlineQueryResultArticle(
str(uuid4()),
title=row["description"],
input_message_content=InputTextMessageContent(
row["price_reply"], parse_mode=telegram.ParseMode.MARKDOWN
),
)
except TypeError:
warning(f"{match} caused error in inline query.")
pass
)
if len(results) == 5:
update.inline_query.answer(results)
update.inline_query.answer(results, cache_time=60 * 60)
info("Inline Command was successful")
return
update.inline_query.answer(results)
@ -568,6 +587,7 @@ def main():
dp.add_handler(CommandHandler("random", rand_pick))
dp.add_handler(CommandHandler("donate", donate))
dp.add_handler(CommandHandler("status", status))
dp.add_handler(CommandHandler("inline", inline_query))
# Charting can be slow so they run async.
dp.add_handler(CommandHandler("intra", intra, run_async=True))

42
cg_Crypto.py
View File

@ -9,7 +9,6 @@ from typing import List, Optional, Tuple
import pandas as pd
import requests as r
import schedule
from fuzzywuzzy import fuzz
from markdownify import markdownify
from Symbol import Coin
@ -74,7 +73,7 @@ class cg_Crypto:
"$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
)
symbols = symbols[["id", "symbol", "name", "description"]]
symbols["type_id"] = "$$" + symbols["id"]
symbols["type_id"] = "$$" + symbols["symbol"]
self.symbol_list = symbols
if return_df:
@ -99,45 +98,6 @@ class cg_Crypto:
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.
Parameters
----------
search : str
String used to search, could be a company name or something close to the companies coin ticker.
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).
"""
schedule.run_pending()
search = search.lower()
try: # https://stackoverflow.com/a/3845776/8774114
return self.searched_symbols[search]
except KeyError:
pass
symbols = self.symbol_list
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(
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"])))
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.

2
requirements.txt
View File

@ -1,8 +1,6 @@
python-telegram-bot==13.5
requests==2.25.1
pandas==1.2.1
fuzzywuzzy==0.18.0
python-Levenshtein==0.12.1
schedule==1.0.0
mplfinance==0.12.7a5
markdownify==0.6.5

56
symbol_router.py
View File

@ -9,7 +9,6 @@ from logging import critical, debug, error, info, warning
import pandas as pd
import schedule
from cachetools import TTLCache, cached
from fuzzywuzzy import fuzz
from cg_Crypto import cg_Crypto
from IEX_Symbol import IEX_Symbol
@ -19,7 +18,6 @@ from Symbol import Coin, Stock, Symbol
class Router:
STOCK_REGEX = "(?:^|[^\\$])\\$([a-zA-Z.]{1,6})"
CRYPTO_REGEX = "[$]{2}([a-zA-Z]{1,20})"
searched_symbols = {}
trending_count = {}
def __init__(self):
@ -110,45 +108,7 @@ class Router:
return stats
def search_symbols(self, search: str) -> list[tuple[str, str]]:
"""Performs a fuzzy search to find stock symbols closest to a search term.
Parameters
----------
search : str
String used to search, could be a company name or something close to the companies stock ticker.
Returns
-------
list[tuple[str, str]]
A list tuples of every stock sorted in order of how well they match.
Each tuple contains: (Symbol, Issue Name).
"""
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
search = search.lower()
df["Match"] = df.apply(
lambda x: fuzz.ratio(search, f"{x['symbol']}".lower()),
axis=1,
)
df.sort_values(by="Match", ascending=False, inplace=True)
# if df["Match"].head().sum() < 300:
# df["Match"] = df.apply(
# lambda x: fuzz.partial_ratio(search, x["name"].lower()),
# axis=1,
# )
# df.sort_values(by="Match", ascending=False, inplace=True)
symbols = df.head(20)
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
self.searched_symbols[search] = symbol_list
return symbol_list
def inline_search(self, search: str) -> list[tuple[str, str]]:
def inline_search(self, search: str, matches: int = 5) -> pd.DataFrame:
"""Searches based on the shortest symbol that contains the same string as the search.
Should be very fast compared to a fuzzy search.
@ -165,16 +125,16 @@ class Router:
df = pd.concat([self.stock.symbol_list, self.crypto.symbol_list])
search = search.lower()
df = df[
df["description"].str.contains(search, regex=False, case=False)
].sort_values(by="type_id", key=lambda x: x.str.len())
df = df[df["type_id"].str.contains(search, regex=False)].sort_values(
by="type_id", key=lambda x: x.str.len()
symbols = df.head(matches)
symbols["price_reply"] = symbols["type_id"].apply(
lambda sym: self.price_reply(self.find_symbols(sym))[0]
)
symbols = df.head(20)
symbol_list = list(zip(list(symbols["symbol"]), list(symbols["description"])))
self.searched_symbols[search] = symbol_list
return symbol_list
return symbols
def price_reply(self, symbols: list[Symbol]) -> list[str]:
"""Returns current market price or after hours if its available for a given stock symbol.