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 '101-combine-discord-and-telegram-bots-into-one-repo-and-deploy-with-docker-compose' into 'master'

Browse Source 
Resolve "Combine Discord and Telegram bots into one repo and deploy with docker compose"

Closes #101

See merge request simple-stock-bots/simple-telegram-stock-bot!49
This commit is contained in:
Anson Biggs 2023-09-03 19:01:09 +00:00
commit 10c9a5a9e4
No known key found for this signature in database
17 changed files with 688 additions and 407 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

0
.env Normal file
View File

3
.gitignore vendored
View File

@ -1 +1,2 @@
__pycache__ __pycache__
.env

31
.gitlab-ci.yml
View File

@ -1,34 +1,5 @@
black: black:
stage: .pre # stage: .pre
image: registry.gitlab.com/pipeline-components/black:latest image: registry.gitlab.com/pipeline-components/black:latest
script: script:
- black --check --verbose -- . - black --check --verbose -- .
build:master:
stage: build
image:
name: gcr.io/kaniko-project/executor:v1.9.0-debug
entrypoint: [""]
script:
- /kaniko/executor
--context "${CI_PROJECT_DIR}"
--dockerfile "${CI_PROJECT_DIR}/Dockerfile"
--destination "${CI_REGISTRY_IMAGE}:latest"
rules:
- if: '$CI_COMMIT_BRANCH == "master"'
build:branch:
stage: build
image:
name: gcr.io/kaniko-project/executor:v1.9.0-debug
entrypoint: [""]
script:
- /kaniko/executor
--context "${CI_PROJECT_DIR}"
--dockerfile "${CI_PROJECT_DIR}/Dockerfile"
--destination "${CI_REGISTRY_IMAGE}:${CI_COMMIT_SHORT_SHA}"
--destination "${CI_REGISTRY_IMAGE}:${CI_COMMIT_BRANCH}"
rules:
- if: '$CI_COMMIT_BRANCH != "master"'

2
MarketData.py → common/MarketData.py
View File

@ -8,7 +8,7 @@ import pytz
import requests as r import requests as r
import schedule import schedule
from Symbol import Stock from common.Symbol import Stock
log = logging.getLogger(__name__) log = logging.getLogger(__name__)

0
Symbol.py → common/Symbol.py
View File

734
cg_Crypto.py → common/cg_Crypto.py
View File

@ -1,367 +1,367 @@
import logging import logging
from typing import List from typing import List
import pandas as pd import pandas as pd
import requests as r import requests as r
import schedule import schedule
from markdownify import markdownify from markdownify import markdownify
from Symbol import Coin from common.Symbol import Coin
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
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
trending_cache: List[str] = [] trending_cache: List[str] = []
def __init__(self) -> None: def __init__(self) -> None:
self.get_symbol_list() self.get_symbol_list()
schedule.every().day.do(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 url = "https://api.coingecko.com/api/v3" + endpoint
resp = r.get(url, params=params, timeout=timeout) resp = r.get(url, params=params, timeout=timeout)
# Make sure API returned a proper status code # Make sure API returned a proper status code
try: try:
resp.raise_for_status() resp.raise_for_status()
except r.exceptions.HTTPError as e: except r.exceptions.HTTPError as e:
log.error(e) log.error(e)
return {} return {}
# Make sure API returned valid JSON # Make sure API returned valid JSON
try: try:
resp_json = resp.json() resp_json = resp.json()
return resp_json return resp_json
except r.exceptions.JSONDecodeError as e: except r.exceptions.JSONDecodeError as e:
log.error(e) log.error(e)
return {} return {}
def symbol_id(self, symbol) -> str: def symbol_id(self, symbol) -> str:
try: try:
return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[0] return self.symbol_list[self.symbol_list["symbol"] == symbol]["id"].values[0]
except KeyError: except KeyError:
return "" return ""
def get_symbol_list(self): def get_symbol_list(self):
raw_symbols = self.get("/coins/list") raw_symbols = self.get("/coins/list")
symbols = pd.DataFrame(data=raw_symbols) symbols = pd.DataFrame(data=raw_symbols)
# Removes all binance-peg symbols # Removes all binance-peg symbols
symbols = symbols[~symbols["id"].str.contains("binance-peg")] symbols = symbols[~symbols["id"].str.contains("binance-peg")]
symbols["description"] = "$$" + symbols["symbol"].str.upper() + ": " + symbols["name"] symbols["description"] = "$$" + symbols["symbol"].str.upper() + ": " + symbols["name"]
symbols = symbols[["id", "symbol", "name", "description"]] symbols = symbols[["id", "symbol", "name", "description"]]
symbols["type_id"] = "$$" + symbols["symbol"] symbols["type_id"] = "$$" + symbols["symbol"]
self.symbol_list = symbols self.symbol_list = symbols
def status(self) -> str: def status(self) -> str:
"""Checks CoinGecko /ping endpoint for API issues. """Checks CoinGecko /ping endpoint for API issues.
Returns Returns
------- -------
str str
Human readable text on status of CoinGecko API Human readable text on status of CoinGecko API
""" """
status = r.get( status = r.get(
"https://api.coingecko.com/api/v3/ping", "https://api.coingecko.com/api/v3/ping",
timeout=5, timeout=5,
) )
try: try:
status.raise_for_status() status.raise_for_status()
return ( return (
f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds." f"CoinGecko API responded that it was OK with a {status.status_code} in {status.elapsed.total_seconds()} Seconds."
) )
except r.HTTPError: except r.HTTPError:
return f"CoinGecko API returned an error code {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."
def price_reply(self, coin: Coin) -> str: def price_reply(self, coin: Coin) -> str:
"""Returns current market price or after hours if its available for a given coin symbol. """Returns current market price or after hours if its available for a given coin symbol.
Parameters Parameters
---------- ----------
symbols : list symbols : list
List of coin symbols. List of coin symbols.
Returns Returns
------- -------
Dict[str, str] Dict[str, str]
Each symbol passed in is a key with its value being a human readable Each symbol passed in is a key with its value being a human readable
markdown formatted string of the symbols price and movement. markdown formatted string of the symbols price and movement.
""" """
if resp := self.get( if resp := self.get(
"/simple/price", "/simple/price",
params={ params={
"ids": coin.id, "ids": coin.id,
"vs_currencies": self.vs_currency, "vs_currencies": self.vs_currency,
"include_24hr_change": "true", "include_24hr_change": "true",
}, },
): ):
try: try:
data = resp[coin.id] data = resp[coin.id]
price = data[self.vs_currency] price = data[self.vs_currency]
change = data[self.vs_currency + "_24h_change"] change = data[self.vs_currency + "_24h_change"]
if change is None: if change is None:
change = 0 change = 0
except KeyError: except KeyError:
return f"{coin.id} returned an error." 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 # Determine wording of change text
if change > 0: if change > 0:
message += f", the coin is currently **up {change:.3f}%** for today" message += f", the coin is currently **up {change:.3f}%** for today"
elif change < 0: elif change < 0:
message += f", the coin is currently **down {change:.3f}%** for today" message += f", the coin is currently **down {change:.3f}%** for today"
else: else:
message += ", the coin hasn't shown any movement today." message += ", the coin hasn't shown any movement today."
else: else:
message = f"The price for {coin.name} is not available. If you suspect this is an error run `/status`" 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: def intra_reply(self, symbol: Coin) -> pd.DataFrame:
"""Returns price data for a symbol since the last market open. """Returns price data for a symbol since the last market open.
Parameters Parameters
---------- ----------
symbol : str symbol : str
Stock symbol. Stock symbol.
Returns Returns
------- -------
pd.DataFrame pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
""" """
if resp := self.get( if resp := self.get(
f"/coins/{symbol.id}/ohlc", f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 1}, params={"vs_currency": self.vs_currency, "days": 1},
): ):
df = pd.DataFrame(resp, columns=["Date", "Open", "High", "Low", "Close"]).dropna() df = pd.DataFrame(resp, columns=["Date", "Open", "High", "Low", "Close"]).dropna()
df["Date"] = pd.to_datetime(df["Date"], unit="ms") df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date") df = df.set_index("Date")
return df return df
return pd.DataFrame() return pd.DataFrame()
def chart_reply(self, symbol: Coin) -> 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. """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. Also caches multiple requests made in the same day.
Parameters Parameters
---------- ----------
symbol : str symbol : str
Stock symbol. Stock symbol.
Returns Returns
------- -------
pd.DataFrame pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame. Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
""" """
if resp := self.get( if resp := self.get(
f"/coins/{symbol.id}/ohlc", f"/coins/{symbol.id}/ohlc",
params={"vs_currency": self.vs_currency, "days": 30}, params={"vs_currency": self.vs_currency, "days": 30},
): ):
df = pd.DataFrame(resp, columns=["Date", "Open", "High", "Low", "Close"]).dropna() df = pd.DataFrame(resp, columns=["Date", "Open", "High", "Low", "Close"]).dropna()
df["Date"] = pd.to_datetime(df["Date"], unit="ms") df["Date"] = pd.to_datetime(df["Date"], unit="ms")
df = df.set_index("Date") df = df.set_index("Date")
return df return df
return pd.DataFrame() return pd.DataFrame()
def stat_reply(self, symbol: Coin) -> str: def stat_reply(self, symbol: Coin) -> str:
"""Gathers key statistics on coin. Mostly just CoinGecko scores. """Gathers key statistics on coin. Mostly just CoinGecko scores.
Parameters Parameters
---------- ----------
symbol : Coin symbol : Coin
Returns Returns
------- -------
str str
Preformatted markdown. Preformatted markdown.
""" """
if data := self.get( if data := self.get(
f"/coins/{symbol.id}", f"/coins/{symbol.id}",
params={ params={
"localization": "false", "localization": "false",
}, },
): ):
return f""" return f"""
[{data['name']}]({data['links']['homepage'][0]}) Statistics: [{data['name']}]({data['links']['homepage'][0]}) Statistics:
Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,} Market Cap: ${data['market_data']['market_cap'][self.vs_currency]:,}
Market Cap Ranking: {data.get('market_cap_rank',"Not Available")} Market Cap Ranking: {data.get('market_cap_rank',"Not Available")}
CoinGecko Scores: CoinGecko Scores:
Overall: {data.get('coingecko_score','Not Available')} Overall: {data.get('coingecko_score','Not Available')}
Development: {data.get('developer_score','Not Available')} Development: {data.get('developer_score','Not Available')}
Community: {data.get('community_score','Not Available')} Community: {data.get('community_score','Not Available')}
Public Interest: {data.get('public_interest_score','Not Available')} Public Interest: {data.get('public_interest_score','Not Available')}
""" """
else: else:
return f"{symbol.symbol} returned an error." return f"{symbol.symbol} returned an error."
def cap_reply(self, coin: Coin) -> str: def cap_reply(self, coin: Coin) -> str:
"""Gets market cap for Coin """Gets market cap for Coin
Parameters Parameters
---------- ----------
coin : Coin coin : Coin
Returns Returns
------- -------
str str
Preformatted markdown. Preformatted markdown.
""" """
if resp := self.get( if resp := self.get(
"/simple/price", "/simple/price",
params={ params={
"ids": coin.id, "ids": coin.id,
"vs_currencies": self.vs_currency, "vs_currencies": self.vs_currency,
"include_market_cap": "true", "include_market_cap": "true",
}, },
): ):
log.debug(resp) log.debug(resp)
try: try:
data = resp[coin.id] data = resp[coin.id]
price = data[self.vs_currency] price = data[self.vs_currency]
cap = data[self.vs_currency + "_market_cap"] cap = data[self.vs_currency + "_market_cap"]
except KeyError: except KeyError:
return f"{coin.id} returned an error." return f"{coin.id} returned an error."
if cap == 0: if cap == 0:
return f"The market cap for {coin.name} is not available for unknown reasons." return f"The market cap for {coin.name} is not available for unknown reasons."
message = ( message = (
f"The current price of {coin.name} is $**{price:,}** and" f"The current price of {coin.name} is $**{price:,}** and"
+ " its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}" + " its market cap is $**{cap:,.2f}** {self.vs_currency.upper()}"
) )
else: else:
message = f"The Coin: {coin.name} was not found or returned and error." message = f"The Coin: {coin.name} was not found or returned and error."
return message return message
def info_reply(self, symbol: Coin) -> str: def info_reply(self, symbol: Coin) -> str:
"""Gets coin description """Gets coin description
Parameters Parameters
---------- ----------
symbol : Coin symbol : Coin
Returns Returns
------- -------
str str
Preformatted markdown. Preformatted markdown.
""" """
if data := self.get( if data := self.get(
f"/coins/{symbol.id}", f"/coins/{symbol.id}",
params={"localization": "false"}, params={"localization": "false"},
): ):
try: try:
return markdownify(data["description"]["en"]) return markdownify(data["description"]["en"])
except KeyError: except KeyError:
return f"{symbol} does not have a description available." 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 spark_reply(self, symbol: Coin) -> str: def spark_reply(self, symbol: Coin) -> str:
change = self.get( change = self.get(
"/simple/price", "/simple/price",
params={ params={
"ids": symbol.id, "ids": symbol.id,
"vs_currencies": self.vs_currency, "vs_currencies": self.vs_currency,
"include_24hr_change": "true", "include_24hr_change": "true",
}, },
)[symbol.id]["usd_24h_change"] )[symbol.id]["usd_24h_change"]
return f"`{symbol.tag}`: {symbol.name}, {change:.2f}%" return f"`{symbol.tag}`: {symbol.name}, {change:.2f}%"
def trending(self) -> list[str]: def trending(self) -> list[str]:
"""Gets current coins trending on coingecko """Gets current coins trending on coingecko
Returns Returns
------- -------
list[str] list[str]
list of $$ID: NAME, CHANGE% list of $$ID: NAME, CHANGE%
""" """
coins = self.get("/search/trending") coins = self.get("/search/trending")
try: try:
trending = [] trending = []
for coin in coins["coins"]: for coin in coins["coins"]:
c = coin["item"] c = coin["item"]
sym = c["symbol"].upper() sym = c["symbol"].upper()
name = c["name"] name = c["name"]
change = self.get( change = self.get(
"/simple/price", "/simple/price",
params={ params={
"ids": c["id"], "ids": c["id"],
"vs_currencies": self.vs_currency, "vs_currencies": self.vs_currency,
"include_24hr_change": "true", "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: except Exception as e:
log.warning(e) log.warning(e)
return self.trending_cache return self.trending_cache
self.trending_cache = trending self.trending_cache = trending
return trending return trending
def batch_price(self, coins: list[Coin]) -> list[str]: def batch_price(self, coins: list[Coin]) -> list[str]:
"""Gets price of a list of coins all in one API call """Gets price of a list of coins all in one API call
Parameters Parameters
---------- ----------
coins : list[Coin] coins : list[Coin]
Returns Returns
------- -------
list[str] list[str]
returns preformatted list of strings detailing price movement of each coin passed in. 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( prices = self.get(
"/simple/price", "/simple/price",
params={ params={
"ids": query, "ids": query,
"vs_currencies": self.vs_currency, "vs_currencies": self.vs_currency,
"include_24hr_change": "true", "include_24hr_change": "true",
}, },
) )
replies = [] replies = []
for coin in coins: for coin in coins:
if coin.id in prices: if coin.id in prices:
p = prices[coin.id] p = prices[coin.id]
if p.get("usd_24h_change") is None: if p.get("usd_24h_change") is None:
p["usd_24h_change"] = 0 p["usd_24h_change"] = 0
replies.append( 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." 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

6
symbol_router.py → common/symbol_router.py
View File

@ -10,9 +10,9 @@ import pandas as pd
import schedule import schedule
from cachetools import TTLCache, cached from cachetools import TTLCache, cached
from cg_Crypto import cg_Crypto from common.cg_Crypto import cg_Crypto
from MarketData import MarketData from common.MarketData import MarketData
from Symbol import Coin, Stock, Symbol from common.Symbol import Coin, Stock, Symbol
from typing import Dict from typing import Dict

2
dev-reqs.txt
View File

@ -1,4 +1,4 @@
-r requirements.txt -r telegram/requirements.txt
black==23.3.0 black==23.3.0
flake8==5.0.4 flake8==5.0.4
Flake8-pyproject==1.2.3 Flake8-pyproject==1.2.3

65
discord/D_info.py Normal file
View File

@ -0,0 +1,65 @@
"""Functions and Info specific to the discord Bot
"""
import re
import requests as r
class D_info:
license = re.sub(
r"\b\n",
" ",
r.get("https://gitlab.com/simple-stock-bots/simple-discord-stock-bot/-/raw/master/LICENSE").text,
)
help_text = """
Thanks for using this bot, consider supporting it by [buying me a beer.](https://www.buymeacoffee.com/Anson)
If you are interested in stock market data, or want to host your own bot, be sure to use my affiliate link so that the bot can stay free: [marketdata.app](https://dashboard.marketdata.app/marketdata/aff/go/misterbiggs?keyword=discord)
Keep up with the latest news for the bot in its discord Channel: https://t.me/simplestockbotnews
Full documentation on using and running your own stock bot can be found on the bots [docs.](https://docs.simplestockbot.com)
The bot detects _"Symbols"_ using either one `$` or two `$$` dollar signs before the symbol. One dollar sign is for a stock market ticker, while two is for a cryptocurrency coin. `/chart $$eth` would return a chart of the past month of data for Ethereum, while `/dividend $psec` returns dividend information for Prospect Capital stock.
Simply calling a symbol in any message that the bot can see will also return the price. So a message like: `I wonder if $$btc will go to the Moon now that $tsla accepts it as payment` would return the current price for both Bitcoin and Tesla.
**Commands**
- `/donate [amount in USD]` to donate. 🎗️
- `/intra $[symbol]` Plot of the stocks movement since the last market open. 📈
- `/chart $[symbol]` Plot of the stocks movement for the past 1 month. 📊
- `/trending` Trending Stocks and Cryptos. 💬
- `/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 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 [marketdata.app](https://dashboard.marketdata.app/marketdata/aff/go/misterbiggs?keyword=discord)
If you believe the bot is not behaving properly run `/status` or [get in touch](https://docs.simplestockbot.com/contact).
"""
donate_text = """
Simple Stock Bot is run entirely on donations[.](https://www.buymeacoffee.com/Anson)
All donations go directly towards paying for servers, and market data is provided by
[marketdata.app](https://dashboard.marketdata.app/marketdata/aff/go/misterbiggs?keyword=discord).
The easiest way to donate is to run the `/donate [amount in USD]` command with US dollars you would like to donate.
Example: `/donate 2` would donate 2 USD.
An alternative way to donate is through https://www.buymeacoffee.com/Anson which requires no account and accepts Paypal or Credit card.
If you have any questions see the [website](https://docs.simplestockbot.com)
"""
commands = """ # Not used by the bot but for updating commands with BotFather
donate - Donate to the bot 🎗️
help - Get some help using the bot. 🆘
trending - Trending Stocks and Cryptos. 💬
intra - $[symbol] Plot since the last market open. 📈
chart - $[chart] Plot of the past month. 📊
"""

7
Dockerfile → discord/Dockerfile
View File

@ -1,7 +1,8 @@
FROM python:3.11-buster AS builder FROM python:3.11-buster AS builder
COPY requirements.txt /requirements.txt COPY discord/requirements.txt .
RUN pip install --user -r requirements.txt RUN pip install --user -r requirements.txt
@ -11,7 +12,7 @@ ENV MPLBACKEND=Agg
COPY --from=builder /root/.local /root/.local COPY --from=builder /root/.local /root/.local
COPY common common
COPY . . COPY discord .
CMD [ "python", "./bot.py" ] CMD [ "python", "./bot.py" ]

202
discord/bot.py Normal file
View File

@ -0,0 +1,202 @@
import datetime
import io
import logging
import os
import mplfinance as mpf
import nextcord
from nextcord.ext import commands
from D_info import D_info
from common.symbol_router import Router
DISCORD_TOKEN = os.environ["DISCORD"]
s = Router()
d = D_info()
intents = nextcord.Intents.default()
client = nextcord.Client(intents=intents)
bot = commands.Bot(command_prefix="/", description=d.help_text, intents=intents)
logger = logging.getLogger("nextcord")
logger.setLevel(logging.INFO)
handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8", mode="w")
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
logger.addHandler(handler)
@bot.event
async def on_ready():
logging.info("Starting Simple Stock Bot")
logging.info(f"Logged in as {bot.user.name} {bot.user.id}")
@bot.command()
async def status(ctx: commands):
"""Debug command for diagnosing if the bot is experiencing any issues."""
logging.warning(f"Status command ran by {ctx.message.author}")
message = ""
try:
message = "Contact MisterBiggs#0465 if you need help.\n"
message += s.status(f"Bot recieved your message in: {bot.latency*1000:.4f}ms") + "\n"
except Exception as ex:
logging.critical(ex)
message += (
f"*\n\nERROR ENCOUNTERED:*\n{ex}\n\n"
+ "*The bot encountered an error while attempting to find errors. Please contact the bot admin.*"
)
await ctx.send(message)
@bot.command()
async def license(ctx: commands):
"""Returns the bots license agreement."""
await ctx.send(d.license)
@bot.command()
async def donate(ctx: commands):
"""Details on how to support the development and hosting of the bot."""
await ctx.send(d.donate_text)
@bot.command()
async def search(ctx: commands, *, query: str):
"""Search for a stock symbol using either symbol of company name."""
results = s.search_symbols(query)
if results:
reply = "*Search Results:*\n`$ticker: Company Name`\n"
for query in results:
reply += "`" + query[1] + "`\n"
await ctx.send(reply)
@bot.command()
async def crypto(ctx: commands, _: str):
"""Get the price of a cryptocurrency using in USD."""
await ctx.send("Crypto now has native support. Any crypto can be called using two dollar signs: `$$eth` `$$btc` `$$doge`")
@bot.command()
async def intra(ctx: commands, sym: str):
"""Get a chart for the stocks movement since market open."""
symbols = s.find_symbols(sym)
if len(symbols):
symbol = symbols[0]
else:
await ctx.send("No symbols or coins found.")
return
df = s.intra_reply(symbol)
if df.empty:
await ctx.send("Invalid symbol please see `/help` for usage details.")
return
with ctx.channel.typing():
buf = io.BytesIO()
mpf.plot(
df,
type="renko",
title=f"\n{symbol.name}",
volume="volume" in df.keys(),
style="yahoo",
savefig=dict(fname=buf, dpi=400, bbox_inches="tight"),
)
buf.seek(0)
# Get price so theres no request lag after the image is sent
price_reply = s.price_reply([symbol])[0]
await ctx.send(
file=nextcord.File(
buf,
filename=f"{symbol.name}:intra{datetime.date.today().strftime('%S%M%d%b%Y')}.png",
),
content=f"\nIntraday chart for {symbol.name} from {df.first_valid_index().strftime('%d %b at %H:%M')} to"
+ f" {df.last_valid_index().strftime('%d %b at %H:%M')}",
)
await ctx.send(price_reply)
@bot.command()
async def chart(ctx: commands, sym: str):
"""returns a chart of the past month of data for a symbol"""
symbols = s.find_symbols(sym)
if len(symbols):
symbol = symbols[0]
else:
await ctx.send("No symbols or coins found.")
return
df = s.chart_reply(symbol)
if df.empty:
await ctx.send("Invalid symbol please see `/help` for usage details.")
return
with ctx.channel.typing():
buf = io.BytesIO()
mpf.plot(
df,
type="candle",
title=f"\n{symbol.name}",
volume="volume" in df.keys(),
style="yahoo",
savefig=dict(fname=buf, dpi=400, bbox_inches="tight"),
)
buf.seek(0)
# Get price so theres no request lag after the image is sent
price_reply = s.price_reply([symbol])[0]
await ctx.send(
file=nextcord.File(
buf,
filename=f"{symbol.name}:1M{datetime.date.today().strftime('%d%b%Y')}.png",
),
content=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')}",
)
await ctx.send(price_reply)
@bot.command()
async def cap(ctx: commands, sym: str):
"""Get the market cap of a symbol"""
symbols = s.find_symbols(sym)
if symbols:
with ctx.channel.typing():
for reply in s.cap_reply(symbols):
await ctx.send(reply)
@bot.command()
async def trending(ctx: commands):
"""Get a list of Trending Stocks and Coins"""
with ctx.channel.typing():
await ctx.send(s.trending())
@bot.event
async def on_message(message):
if message.author.id == bot.user.id:
return
if message.content:
if message.content[0] == "/":
await bot.process_commands(message)
return
if "$" in message.content:
symbols = s.find_symbols(message.content)
if symbols:
for reply in s.price_reply(symbols):
await message.channel.send(reply)
return
bot.run(DISCORD_TOKEN)

7
discord/requirements.txt Normal file
View File

@ -0,0 +1,7 @@
nextcord==2.4.2
requests==2.25.1
pandas==2.0.0
schedule==1.0.0
mplfinance==0.12.7a5
markdownify==0.6.5
cachetools==4.2.2

13
docker-compose.yaml Normal file
View File

@ -0,0 +1,13 @@
version: '3'
services:
telegram:
build:
context: .
dockerfile: telegram/Dockerfile
image: registry.gitlab.com/simple-stock-bots/simple-telegram-stock-bot
env_file: .env
discord:
build:
context: .
dockerfile: discord/Dockerfile
env_file: .env

21
telegram/Dockerfile Normal file
View File

@ -0,0 +1,21 @@
FROM python:3.11-buster AS builder
COPY telegram/requirements.txt .
RUN pip install --user -r requirements.txt
FROM python:3.11-slim
ENV MPLBACKEND=Agg
COPY --from=builder /root/.local /root/.local
COPY common common
COPY telegram .
CMD [ "python", "./bot.py" ]

0
T_info.py → telegram/T_info.py
View File

2
bot.py → telegram/bot.py
View File

@ -28,7 +28,7 @@ from telegram.ext import (
Updater, Updater,
) )
from symbol_router import Router from common.symbol_router import Router
from T_info import T_info from T_info import T_info
# Enable logging # Enable logging

0
requirements.txt → telegram/requirements.txt
View File