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

fixed inline #82

Browse Source
This commit is contained in:
Anson Biggs 2021-11-05 23:28:44 -07:00
parent 2f3cef4ec4
commit 87aeafc77f
5 changed files with 31 additions and 156 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
IEX_Symbol.py
View File

@ -10,7 +10,6 @@ from typing import List, Optional, Tuple
import pandas as pd
import requests as r
import schedule
from fuzzywuzzy import fuzz
from Symbol import Stock
@ -141,46 +140,6 @@ class IEX_Symbol:
+ " Please check the status page for more information. https://status.iexapis.com"
)
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).
"""
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, symbol: Stock) -> str:
"""Returns price movement of Stock for the last market day, or after hours.

34
bot.py
View File

@ -261,8 +261,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 +275,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"
print(reply)
update.message.reply_text(
text=reply,
parse_mode=telegram.ParseMode.MARKDOWN,
@ -466,28 +468,23 @@ 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))
matches = s.inline_search(update.inline_query.query)
results = []
for match, price in zip(matches, prices):
try:
for _, row in matches.iterrows():
print(row)
results.append(
InlineQueryResultArticle(
str(uuid4()),
title=match[1],
title=row["description"],
input_message_content=InputTextMessageContent(
price, parse_mode=telegram.ParseMode.MARKDOWN
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)
@ -568,6 +565,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.