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

get bot working with marketdata api

Browse Source
This commit is contained in:
Anson Biggs 2023-04-06 23:27:02 -06:00
parent bc1ad95c75
commit b72f0518c2
6 changed files with 91 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.vscode/launch.json vendored
View File

@ -1,15 +1,12 @@
{
// 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"
"console": "integratedTerminal",
"python": "python3.11"
}
]
}

134
MarketData.py
View File

@ -3,7 +3,7 @@
import logging
import os
from datetime import datetime
import datetime as dt
from logging import warning
from typing import List, Optional, Tuple
@ -35,7 +35,7 @@ class MarketData:
MarketData.app API Token
"""
try:
self.MARKETDATA_TOKEN = os.environ["MarketData"]
self.MARKETDATA_TOKEN = os.environ["MARKETDATA"]
if self.MARKETDATA_TOKEN == "TOKEN":
self.MARKETDATA_TOKEN = ""
@ -46,12 +46,9 @@ class MarketData:
)
if self.MARKETDATA_TOKEN != "":
self.get_symbol_list()
schedule.every().day.do(self.get_symbol_list)
schedule.every().day.do(self.clear_charts)
def get(self, endpoint, params: dict = {}, timeout=5) -> dict:
def get(self, endpoint, params: dict = {}, timeout=10) -> dict:
url = "https://api.marketdata.app/v1/" + endpoint
# set token param if it wasn't passed.
@ -91,43 +88,6 @@ class MarketData:
"""
self.charts = {}
def get_symbol_list(
self, return_df=False
) -> Optional[Tuple[pd.DataFrame, datetime]]:
"""Gets list of all symbols supported by IEX
Parameters
----------
return_df : bool, optional
return the dataframe of all stock symbols, by default False
Returns
-------
Optional[Tuple[pd.DataFrame, datetime]]
If `return_df` is set to `True` returns a dataframe, otherwise returns `None`.
"""
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)
self.otc_list = set(otc["symbol"].to_list())
symbols = pd.concat([reg, otc])
# IEX uses backtick ` as apostrophe which breaks telegram markdown parsing
symbols["name"] = symbols["name"].str.replace("`", "'")
symbols["description"] = "$" + symbols["symbol"] + ": " + symbols["name"]
symbols["id"] = symbols["symbol"]
symbols["type_id"] = "$" + symbols["symbol"].str.lower()
symbols = symbols[["id", "symbol", "name", "description", "type_id"]]
self.symbol_list = symbols
if return_df:
return symbols, datetime.now()
def status(self) -> str:
return "status isnt implemented by marketdata.app"
@ -144,12 +104,64 @@ class MarketData:
Formatted markdown
"""
if quoteResp := self.get(f"/stocks/quotes/{symbol}/"):
if quoteResp := self.get(f"stocks/quotes/{symbol}/"):
return f"The current price of {quoteResp['symbol']} is ${quoteResp['last']}"
else:
return f"Getting a quote for {symbol} encountered an error."
def intra_reply(self, symbol: Stock) -> 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.
Parameters
----------
symbol : str
Stock symbol.
Returns
-------
pd.DataFrame
Returns a timeseries dataframe with high, low, and volume data if its available. Otherwise returns empty pd.DataFrame.
"""
schedule.run_pending()
try:
return self.charts[symbol.id.upper()]
except KeyError:
pass
to_date = dt.datetime.today().strftime("%Y-%m-%d")
resolution = "5" # minutes
if data := self.get(
f"stocks/candles/{resolution}/{symbol}",
params={
"from": dt.datetime.now().strftime("%Y-%m-%d"),
"to": dt.datetime.now().isoformat(),
},
):
data.pop("s")
df = pd.DataFrame(data)
df["t"] = pd.to_datetime(df["t"], unit="s")
df.set_index("t", inplace=True)
df.rename(
columns={
"o": "Open",
"h": "High",
"l": "Low",
"c": "Close",
"v": "Volume",
},
inplace=True,
)
self.charts[symbol.id.upper()] = df
return df
return pd.DataFrame()
def chart_reply(self, symbol: Stock) -> 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.
@ -171,24 +183,34 @@ class MarketData:
except KeyError:
pass
to_date = datetime.today().strftime("%Y-%m-%d")
from_date = (datetime.today() - datetime.timedelta(days=30)).strftime(
"%Y-%m-%d"
)
to_date = dt.datetime.today().strftime("%Y-%m-%d")
from_date = (dt.datetime.today() - dt.timedelta(days=30)).strftime("%Y-%m-%d")
resultion = "daily"
if data := self.get(
f"/stocks/candles/h/{symbol}",
f"stocks/candles/{resultion}/{symbol}",
params={
"chartInterval": 3,
"to": to_date,
"from": from_date,
"to": to_date,
},
):
df = pd.DataFrame(
data, columns=["Date", "Open", "High", "Low", "Close"]
).dropna()
df["DT"] = pd.to_datetime(df["date"] + "T" + df["minute"])
df = df.set_index("DT")
data.pop("s")
df = pd.DataFrame(data)
df["t"] = pd.to_datetime(df["t"], unit="s")
df.set_index("t", inplace=True)
df.rename(
columns={
"o": "Open",
"h": "High",
"l": "Low",
"c": "Close",
"v": "Volume",
},
inplace=True,
)
self.charts[symbol.id.upper()] = df
return df

16
Symbol.py
View File

@ -27,17 +27,13 @@ class Symbol:
class Stock(Symbol):
"""Stock Market Object. Gets data from IEX Cloud"""
"""Stock Market Object. Gets data from MarketData"""
def __init__(self, symbol: pd.DataFrame) -> None:
if len(symbol) > 1:
logging.info(f"Crypto with shared id:\n\t{symbol.id}")
symbol = symbol.head(1)
self.symbol = symbol.symbol.values[0]
self.id = symbol.id.values[0]
self.name = symbol.name.values[0]
self.tag = symbol.type_id.values[0].upper()
def __init__(self, symbol: str) -> None:
self.symbol = symbol
self.id = symbol
self.name = "$" + symbol
self.tag = "$" + symbol.upper()
class Coin(Symbol):

4
bot.py
View File

@ -238,7 +238,7 @@ def intra(update: Update, context: CallbackContext):
df,
type="renko",
title=f"\n{symbol.name}",
volume="volume" in df.keys(),
volume="Volume" in df.keys(),
style="yahoo",
savefig=dict(fname=buf, dpi=400, bbox_inches="tight"),
)
@ -292,7 +292,7 @@ def chart(update: Update, context: CallbackContext):
df,
type="candle",
title=f"\n{symbol.name}",
volume="volume" in df.keys(),
volume="Volume" in df.keys(),
style="yahoo",
savefig=dict(fname=buf, dpi=400, bbox_inches="tight"),
)

2
requirements.txt
View File

@ -1,6 +1,6 @@
python-telegram-bot==13.5
requests==2.25.1
pandas==1.2.1
pandas==2.0.0
schedule==1.0.0
mplfinance==0.12.7a5
markdownify==0.6.5

64
symbol_router.py
View File

@ -64,14 +64,8 @@ class Router:
symbols = []
stocks = set(re.findall(self.STOCK_REGEX, text))
for stock in stocks:
sym = self.stock.symbol_list[
self.stock.symbol_list["symbol"].str.fullmatch(stock, case=False)
]
if ~sym.empty:
print(sym)
symbols.append(Stock(sym))
else:
info(f"{stock} is not in list of stocks")
# Market data lacks tools to check if a symbol is valid.
symbols.append(Stock(stock))
coins = set(re.findall(self.CRYPTO_REGEX, text))
for coin in coins:
@ -172,60 +166,6 @@ class Router:
return replies
def dividend_reply(self, symbols: list) -> list[str]:
"""Returns the most recent, or next dividend date for a stock symbol.
Parameters
----------
symbols : list
List of stock symbols.
Returns
-------
Dict[str, str]
Each symbol passed in is a key with its value being a human readable
formatted string of the symbols div dates.
"""
replies = []
for symbol in symbols:
if isinstance(symbol, Stock):
replies.append(self.stock.dividend_reply(symbol))
elif isinstance(symbol, Coin):
replies.append("Cryptocurrencies do no have Dividends.")
else:
debug(f"{symbol} is not a Stock or Coin")
return replies
def news_reply(self, symbols: list) -> list[str]:
"""Gets recent english news on stock symbols.
Parameters
----------
symbols : list
List of stock symbols.
Returns
-------
Dict[str, str]
Each symbol passed in is a key with its value being a human
readable markdown formatted string of the symbols news.
"""
replies = []
for symbol in symbols:
if isinstance(symbol, Stock):
replies.append(self.stock.news_reply(symbol))
elif isinstance(symbol, Coin):
# replies.append(self.crypto.news_reply(symbol))
replies.append(
"News is not yet supported for cryptocurrencies. If you have any suggestions for news sources please contatct @MisterBiggs"
)
else:
debug(f"{symbol} is not a Stock or Coin")
return replies
def info_reply(self, symbols: list) -> list[str]:
"""Gets information on stock symbols.