Scraper#
Module richy.core.scraper contains classes responsible for actuall
web content scraping.
Manager#
Class Manager handles all the methods for scraping actuall items:
class Manager:
"""
Main manager class for scraping.
All scraping methods are placed here.
"""
mobile_user_agent = (
"Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N)"
"AppleWebKit/537.36 (KHTML, like Gecko)"
"Chrome/58.0.3029.33 Mobile Safari/537.36"
)
desktop_user_agent = (
"Mozilla/5.0 (X11; Linux x86_64; rv:77.0) Gecko/20100101 Firefox/77.0"
)
def __repr__(self):
from pprint import pformat
return "<" + type(self).__name__ + "> " + pformat(vars(self), indent=4)
@staticmethod
def get_share_basic_info(share):
"""
Fetches share basic info via rug library.
:param Share share: Share model instance.
:return: Basic info as a dict.
:rtype: dict
"""
api = TipRanks(share.symbol)
try:
return api.get_basic_info()
except SymbolNotFound:
try:
info = yf.Ticker(share.symbol).get_info()
return {
"company_name": info["shortName"],
"market": "",
"description": info["longBusinessSummary"],
"market_cap": info["marketCap"],
"has_dividends": bool(info["dividendYield"]),
"yoy_change": autofloatformat(
info["52WeekChange"] * 100, no_str=True
),
"year_low": info["fiftyTwoWeekLow"],
"year_high": info["fiftyTwoWeekHigh"],
"pe_ratio": info["forwardPE"],
"eps": info["forwardEps"],
"similar_stocks": [],
}
except:
pass
LOGGER.exception(f"Basic info wasn't downloaded", extra={"share": share})
except:
LOGGER.exception("Basic info wasn't downloaded", extra={"share": share})
return {}
@staticmethod
def get_etf_basic_info(etf):
"""
Fetches etf basic info via rug library.
:param Etf etf: Etf model instance.
:return: Basic info as a dict.
:rtype: dict
"""
api = EtfDb(etf.symbol)
try:
return api.get_basic_info()
except SymbolNotFound:
LOGGER.debug(f"Holdings weren't downloaded - symbol {etf} wasn't found.")
except:
LOGGER.exception("Holdings weren't downloaded", extra={"etf": etf})
return {}
@staticmethod
def get_coin_basic_info(coin):
"""
Fetches coin basic info via karpet library.
:param Coin coin: Coin model instance.
:return: Basic info as a dict.
:rtype: dict
"""
api = Karpet()
if coin.coin_id:
return api.get_basic_info(slug=coin.coin_id)
return api.get_basic_info(symbol=coin.symbol)
@staticmethod
def get_dividends(share):
"""
Fetches share dividends via rug library.
:param Share share: Share model instance.
:return: Dividends as a list.
:rtype: list
"""
api = TipRanks(share)
try:
return api.get_dividends()
except SymbolNotFound:
LOGGER.debug(f"Dividends weren't downloaded - symbol {share} wasn't found.")
except:
LOGGER.exception(
"Dividends weren't downloaded",
extra={"share": share},
)
return []
@staticmethod
def get_current_price_and_change(item):
"""
Fetches current market price, market staten and price change in
value and percents.
:param Item item: Item model instance to be fetched price for.
:return: Dataclass with price, state and change values.
:rtype: CurrentPrice
"""
def for_share_or_index_or_etf(symbol):
"""
Note: Yahoo doesn't provide pre/post market prices to indexes.
"""
price = Yahoo(symbol).get_current_price_change()
price_key = {
"pre-market": "pre_market",
"open": "current_market",
"closed": "post_market",
"post-market": "post_market",
}[price["state"]]
return CurrentPrice(
price=float(price[price_key]["value"]),
state=price["state"],
change_value=price[price_key]["change"]["value"],
change_percents=price[price_key]["change"]["percents"],
)
def for_coin(coin):
if coin.coin_id:
kwargs = {"slug": coin.coin_id}
else:
kwargs = {"symbol": coin.symbol}
data = Karpet().get_basic_info(**kwargs)
return CurrentPrice(
price=float(data["current_price"]),
state="open",
change_value=data["price_change_24"],
change_percents=data["price_change_24_percents"],
)
if item.is_coin():
return for_coin(item.coin)
if item.is_share() or item.is_etf():
return for_share_or_index_or_etf(item.symbol)
if item.is_index():
return for_share_or_index_or_etf(f"^{item.symbol}")
@staticmethod
def fetch_price_ratings(share):
"""
Fetches share price ratings data and directly updates them
in the database for the given share.
:param Share share: Share which financials will be downloaded for.
"""
from .models import Asset
fv = FinViz(share.symbol)
try:
Asset.objects.update_or_create(
item=share,
type=Asset.PRICE_RATINGS,
defaults={"data": fv.get_price_ratings()},
)
except SymbolNotFound:
LOGGER.debug(
f"Price ratings weren't downloaded - symbol {share} wasn't found."
)
except:
LOGGER.exception(
"Price ratings weren't downloadeded", extra={"share": share}
)
@staticmethod
def fetch_financials(share):
"""
Fetches all the share financials data and directly
updates them in the database for the given share.
:param Share share: Share which financials will be downloaded for.
"""
from .models import Asset
query = AlphaQuery(share.symbol)
# Revenues.
try:
Asset.objects.update_or_create(
item=share,
type=Asset.REVENUES_DATA,
defaults={"data": query.get_revenues()},
)
LOGGER.debug(f"Revenues for {share} has been downloaded")
except SymbolNotFound:
LOGGER.debug(f"Revenues weren't downloaded - symbol {share} wasn't found.")
except:
LOGGER.exception(
"Revenues weren't downloaded",
extra={"share": share},
)
# Earnings.
try:
Asset.objects.update_or_create(
item=share,
type=Asset.EARNINGS_DATA,
defaults={"data": query.get_earnings()},
)
LOGGER.debug(f"Earnings for {share} has been downloaded")
except SymbolNotFound:
LOGGER.debug(f"Earnings weren't downloaded - symbol {share} wasn't found.")
except:
LOGGER.exception(
"Earnings weren't downloaded",
extra={"share": share},
)
# EPS.
try:
Asset.objects.update_or_create(
item=share, type=Asset.EPS_DATA, defaults={"data": query.get_eps()}
)
LOGGER.debug(f"EPS for {share} has been downloaded")
except SymbolNotFound:
LOGGER.debug(f"EPS weren't downloaded - symbol {share} wasn't found.")
except:
LOGGER.exception(
"EPS wasn't downloaded",
extra={"share": share},
)
@staticmethod
def fetch_ratings(share):
from .models import Asset
bar = BarChart(share.symbol)
try:
Asset.objects.update_or_create(
item=share,
type=Asset.RATINGS_DATA,
defaults={"data": bar.get_ratings()},
)
except:
LOGGER.exception("Ratings weren't downloadeded", extra={"share": share})
@staticmethod
def fetch_share_prices(share, history="max"):
"""
Downloads all prices for the share.
Returns dataframe with following columns:
- Date (index)
- Open
- High
- Low
- Close
- Volume
- Dividends
- Stock Splits
:param Share share: Share model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
try:
ticker = yf.Ticker(share.symbol)
df = ticker.history(period=history)
LOGGER.debug(f"Share prices successfully downloaded for {share}.")
except:
LOGGER.exception("Couldn't fetch share prices.")
return pd.DataFrame()
return df
@staticmethod
def fetch_etf_prices(etf):
"""
Downloads all prices for the etf.
Returns dataframe with following columns:
- Date (index)
- Open
- High
- Low
- Close
- Volume
- Dividends
- Stock Splits
:param Etf etf: Etf model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
try:
ticker = yf.Ticker(etf.symbol)
df = ticker.history("max")
LOGGER.debug(f"Etf prices successfully downloaded for {etf}.")
except:
LOGGER.exception("Couldn't fetch etf prices.")
return pd.DataFrame()
return df
@staticmethod
def fetch_index_prices(index):
"""
Downloads all prices for the index.
Returns dataframe with following columns:
- Date (index)
- Open
- High
- Low
- Close
:param Share share: Share model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
try:
ticker = yf.Ticker(f"^{index.symbol}")
df = ticker.history("max")
# Drop 0 value columns.
df = df.drop(["Volume", "Dividends", "Stock Splits"], axis=1)
LOGGER.debug(f"Index prices successfully downloaded for {index}.")
except:
LOGGER.exception("Couldn't fetch index prices.")
return pd.DataFrame()
return df
@staticmethod
def fetch_coin_prices(coin):
"""
Downloads all prices for the coin since settings.COIN_EPOCH.
Returns dataframe with following columns:
- date (index)
- price
- market_cap
- total_volume
:param Coin coin: Coin model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
LOGGER.debug(f"Downloading prices for {coin.symbol}.")
# Try to download historical data.
try:
karpet = Karpet(settings.COIN_EPOCH, date.today())
df = karpet.fetch_crypto_historical_data(coin.symbol, coin.coin_id)
LOGGER.debug(f"Prices sucessfully downloaded for {coin.symbol}.")
except:
LOGGER.exception(f"Couldn't download historical data for {coin.symbol}.")
return pd.DataFrame()
# Sort the dataframe.
df = df.sort_index()
return df
@staticmethod
def fetch_intraday_prices(item):
"""
Fetches market (intraday) data prices for shares, indexes and ETFs.
For coins past 24 hours prices are fetched in 30 minutes interval.
:param Item item: Item model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
def for_share_or_index_or_etf(item):
"""
Downloads all (including pre/post market) prices for the share
in 5 menut intervals.
Returns dataframe with following columns:
- Date (index)
- Open
- High
- Low
- Close
- Volume
- Dividends
- Stock Splits
:param Share share: Share model instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
open = None
close = None
try:
ticker = yf.Ticker(
f"^{item.symbol}" if item.is_index() else item.symbol
)
df = ticker.history("1d", interval="5m")
df_pp = ticker.history("1d", interval="5m", prepost=True)
# Is market open yet?
if len(df) and df.index[0] > df_pp.index[0]:
open = df.index[0]
# Is market closed yet?
if len(df) and df.index[-1] < df_pp.index[-1]:
close = df.index[-1]
LOGGER.debug(
f"Share intraday prices successfully downloaded for {item.symbol}."
)
except:
LOGGER.exception("Couldn't fetch share intraday prices.")
return ()
return df_pp, open, close
@staticmethod
def for_coin(coin):
"""
Fitchis market prices for past 24 hours with 30 minutes interval.
Returns dataframe with following columns:
- date time (index)
- open
- high
- low
- close
:param Coin coin: Coin modil instance we want prices for.
:return: Pandas dataframe.
:rtype: pandas.DataFrame
"""
k = Karpet()
if coin.coin_id:
return k.fetch_crypto_live_data(slug=coin.coin_id)
return k.fetch_crypto_live_data(symbol=coin.symbol)
if item.is_coin():
return for_coin(item.coin), None, None
if item.is_share() or item.is_index() or item.is_etf():
return for_share_or_index_or_etf(item)
@staticmethod
def fetch_etf_holdings(etf):
"""
:param Item item: Item model instance we want prices for.
"""
api = EtfDb(etf.symbol)
try:
return api.get_holdings()
except SymbolNotFound:
LOGGER.debug(f"Holdings weren't downloaded - symbol {etf} wasn't found.")
except:
LOGGER.exception("Holdings weren't downloaded", extra={"etf": etf})
return {}