API

pstock special

pstock.utils special

chart

get_ohlc_from_chart(data: Dict[str, Any]) -> List[Dict[str, Union[datetime.datetime, float, datetime.timedelta]]]

Source code in pstock/utils/chart.py

def get_ohlc_from_chart(
    data: tp.Dict[str, tp.Any]
) -> tp.List[tp.Dict[str, tp.Union[datetime, float, timedelta]]]:

    result = data.get("chart", {}).get("result")
    if not result:
        error = data.get("chart", {}).get("error")
        if error:
            raise ValueError(f"Yahoo-finance responded with an error:\n{error}")
        raise ValueError(
            "Got invalid value for result field in yahoo-finance chart "
            f"response: {result}"
        )

    result = result[0]
    meta = result["meta"]

    interval = parse_duration(meta["dataGranularity"])
    symbol = meta["symbol"]

    # Empty chart
    if "timestamp" not in result:
        logging.getLogger(__name__).warning(
            f"Yahoo-finance returned an empty chart for symbol '{symbol}'. "
            "Please make sure that provided params are valid (for example that "
            "start/end times are valid UTC market times)."
        )
        return []

    timestamps = result["timestamp"]
    indicators = result["indicators"]
    ohlc = indicators["quote"][0]
    volumes = ohlc["volume"]
    opens = ohlc["open"]
    closes = ohlc["close"]
    lows = ohlc["low"]
    highs = ohlc["high"]

    if "adjclose" in indicators:
        adj_closes = indicators["adjclose"][0]["adjclose"]
    else:
        adj_closes = closes

    return [
        {
            "date": timestamp,
            "close": close if close is not None else np.nan,
            "adj_close": adj_close if adj_close is not None else np.nan,
            "high": high if high is not None else np.nan,
            "low": low if low is not None else np.nan,
            "open": open if open is not None else np.nan,
            "volume": volume if volume is not None else np.nan,
            "interval": interval,
        }
        for timestamp, volume, open, close, adj_close, low, high in zip(
            timestamps, volumes, opens, closes, adj_closes, lows, highs
        )
    ]

financials_quote

get_income_statement_data_from_financials_quote(financials_quote: Dict[str, Any], key: Literal['incomeStatementHistory', 'incomeStatementHistoryQuarterly'] = 'incomeStatementHistory') -> List[Dict[str, Any]]

Source code in pstock/utils/financials_quote.py

def get_income_statement_data_from_financials_quote(
    financials_quote: tp.Dict[str, tp.Any],
    key: tp.Literal[
        "incomeStatementHistory", "incomeStatementHistoryQuarterly"
    ] = "incomeStatementHistory",
) -> tp.List[tp.Dict[str, tp.Any]]:

    statement_history = financials_quote.get(key, {}).get("incomeStatementHistory", [])
    if not statement_history:
        return []

    return [
        {
            "date": statement.get("endDate", {}).get("raw"),
            "ebit": statement.get("ebit", {}).get("raw"),
            "total_revenue": statement.get("totalRevenue", {}).get("raw"),
            "gross_profit": statement.get("grossProfit", {}).get("raw"),
        }
        for statement in statement_history
    ]

quote

get_latest_price_from_quote(price_data: Dict[str, Any]) -> float

Source code in pstock/utils/quote.py

def get_latest_price_from_quote(price_data: tp.Dict[str, tp.Any]) -> float:
    if not price_data:
        raise ValueError("No price data found.")

    # regular market price
    regular_market_price = price_data["regularMarketPrice"]["raw"]
    regular_market_time = pendulum.from_timestamp(price_data["regularMarketTime"])

    prices = {"regular": (regular_market_time, regular_market_price)}

    # pre-market price
    pre_market_price = (
        price_data.get("preMarketPrice", {}).get("raw")
        if price_data.get("preMarketPrice", {}) is not None
        else None
    )
    if pre_market_price is not None:
        prices["pre"] = (
            pendulum.from_timestamp(price_data["preMarketTime"]),
            pre_market_price,
        )

    # post-market price
    post_market_price = (
        price_data.get("postMarketPrice", {}).get("raw")
        if price_data.get("postMarketPrice", {}) is not None
        else None
    )
    if post_market_price is not None:
        prices["post"] = (
            pendulum.from_timestamp(price_data["postMarketTime"]),
            post_market_price,
        )

    _, (_, price) = min(prices.items(), key=lambda x: abs(pendulum.now() - x[1][0]))

    return price

get_asset_data_from_quote(quote: Dict[str, Any]) -> Dict[str, Any]

Source code in pstock/utils/quote.py

def get_asset_data_from_quote(quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
    profile = quote.get("summaryProfile", {}) or {}
    quote_type = quote.get("quoteType", {}) or {}
    price_data = quote.get("price", {}) or {}

    symbol = quote.get("symbol") or quote_type.get("symbol") or price_data.get("symbol")
    name = quote_type.get("longName", price_data.get("longName")) or quote_type.get(
        "shortName", price_data.get("shortName")
    )
    asset_type = quote_type.get("quoteType") or price_data.get("quoteType")
    currency = price_data.get("currency")
    latest_price = get_latest_price_from_quote(price_data)
    sector = profile.get("sector")
    industry = profile.get("industry")

    return {
        "symbol": symbol,
        "name": name,
        "asset_type": asset_type,
        "currency": currency,
        "latest_price": latest_price,
        "sector": sector,
        "industry": industry,
    }

get_earnings_data_from_quote(quote: Dict[str, Any]) -> List[Dict[str, Union[str, float]]]

Source code in pstock/utils/quote.py

def get_earnings_data_from_quote(
    quote: tp.Dict[str, tp.Any]
) -> tp.List[tp.Dict[str, tp.Union[str, float]]]:

    earnings = quote.get("earnings")
    if not earnings or not isinstance(earnings, dict):
        return []

    earnings_chart = earnings.get("earningsChart", {})
    quarterly_earnings = earnings_chart.get("quarterly", [])
    quarterly_financial_chart = earnings.get("financialsChart", {}).get("quarterly", [])

    date_to_earnings = {
        e.get("date", ""): {
            "actual": e.get("actual", {}).get("raw", np.nan),
            "estimate": e.get("estimate", {}).get("raw", np.nan),
        }
        for e in quarterly_earnings
        if "date" in e
    }
    date_to_fin_chart = {
        c.get("date", ""): {
            "revenue": c.get("revenue", {}).get("raw", np.nan),
            "earnings": c.get("earnings", {}).get("raw", np.nan),
        }
        for c in quarterly_financial_chart
        if "date" in c
    }
    all_dates = set(list(date_to_earnings.keys()) + list(date_to_fin_chart.keys()))
    passed_earnings = [
        dict(
            quarter=quarter,
            actual=date_to_earnings.get(quarter, {}).get("actual", np.nan),
            estimate=date_to_earnings.get(quarter, {}).get("estimate", np.nan),
            revenue=date_to_fin_chart.get(quarter, {}).get("revenue", np.nan),
            earnings=date_to_fin_chart.get(quarter, {}).get("earnings", np.nan),
        )
        for quarter in all_dates
    ]

    next_earnings = [
        dict(
            quarter=(
                f"{earnings_chart.get('currentQuarterEstimateDate', '')}"
                f"{earnings_chart.get('currentQuarterEstimateYear', '')}"
            ),
            estimate=earnings_chart.get("currentQuarterEstimate", {}).get(
                "raw", np.nan
            ),
            actual=np.nan,
            revenue=np.nan,
            earnings=np.nan,
        )
    ]
    return passed_earnings + next_earnings

get_trends_data_from_quote(quote: Dict[str, Any]) -> List[Dict[str, Any]]

Source code in pstock/utils/quote.py

def get_trends_data_from_quote(
    quote: tp.Dict[str, tp.Any]
) -> tp.List[tp.Dict[str, tp.Any]]:

    if not quote:
        return []
    recommendation_trend = quote.get("recommendationTrend", {})
    if not recommendation_trend:
        return []
    trends = recommendation_trend.get("trend", [])
    return [
        {
            "date": date.today() + pendulum.duration(months=int(trend["period"][:-1])),
            "strong_buy": trend.get("strongBuy", 0),
            "buy": trend.get("buy", 0),
            "hold": trend.get("hold", 0),
            "sell": trend.get("sell", 0),
            "strong_sell": trend.get("stronSell", 0),
        }
        for trend in trends
    ]

utils

rdm_user_agent_value() -> str

Source code in pstock/utils/utils.py

def rdm_user_agent_value() -> str:
    return random.choice(_USER_AGENT_LIST)

parse_datetime(value: Union[str, int, float, datetime.date, datetime.datetime]) -> DateTime

Source code in pstock/utils/utils.py

def parse_datetime(
    value: tp.Union[str, int, float, date, datetime]
) -> pendulum.DateTime:
    errors: tp.List[str] = []

    # try to parse a datetime string, int, bytes
    if not isinstance(value, date):
        try:
            return pendulum.instance(parse_datetime_pydantic(value))
        except (DateTimeError, TypeError) as error:
            errors.append(str(error))

    # if not above, maybe try to parse a date string, int, bytes object.
    try:
        _date = parse_date_pydantic(value)
        return pendulum.datetime(_date.year, _date.month, _date.day)
    except DateError as error:
        errors.append(str(error))
    raise ValueError(f"Couldn't parse to datetime: {value}: {', '.join(errors)}")

parse_duration(value: Union[str, int, float, datetime.timedelta]) -> Duration

Source code in pstock/utils/utils.py

def parse_duration(value: tp.Union[str, int, float, timedelta]) -> pendulum.Duration:
    if isinstance(value, timedelta):
        return pendulum.duration(seconds=value.total_seconds())

    try:
        return pendulum.duration(seconds=parse_duration_pydantic(value).total_seconds())
    except DurationError as error:
        assert isinstance(value, str), str(error)

    if value.lower() == "mtd":
        return pendulum.duration(months=1)
    if value.lower() == "ytd":
        return pendulum.duration(years=1)

    kwargs: tp.Dict[str, float] = {}
    for match in re.finditer(_UNITS_REGEX, value, flags=re.I):
        unit = match.group("unit").lower()
        val = match.group("val")
        if unit not in _UNITS:
            raise DurationError()
        name, _type = _UNITS[unit]
        if name in kwargs:
            kwargs[name] += _type(val)
        else:
            kwargs[name] = _type(val)

    if not kwargs:
        raise DurationError()
    return pendulum.duration(**kwargs)

httpx_client_manager(client: Optional[httpx.AsyncClient] = None) -> AsyncGenerator[httpx.AsyncClient, NoneType]

Source code in pstock/utils/utils.py

@asynccontextmanager
async def httpx_client_manager(
    client: tp.Optional[httpx.AsyncClient] = None,
) -> tp.AsyncGenerator[httpx.AsyncClient, None]:
    _close = False
    if client is None:
        _close = True
        client = httpx.AsyncClient()
    try:
        yield client
    finally:
        if _close:
            await client.aclose()

pstock.asset

Asset (QuoteSummary) pydantic-model

Source code in pstock/asset.py

class Asset(QuoteSummary):
    symbol: str
    name: str
    asset_type: tp.Literal[
        "EQUITY", "CURRENCY", "CRYPTOCURRENCY", "ETF", "FUTURE", "INDEX"
    ]
    currency: str
    latest_price: float = np.nan
    sector: tp.Optional[str]
    industry: tp.Optional[str]
    earnings: Earnings = Field(repr=False)
    trends: Trends = Field(repr=False)
    income_statement: tp.Optional[IncomeStatements] = Field(repr=False)

    @validator("symbol")
    def symbol_upper(cls, symbol: str) -> str:
        return symbol.upper()

    @classmethod
    def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
        data = get_asset_data_from_quote(quote)
        earnings = Earnings.process_quote(quote)
        trends = Trends.process_quote(quote)
        return {**data, "earnings": earnings, "trends": trends}

    @classmethod
    def process_financials_quote(
        cls, financials_quote: tp.Dict[str, tp.Any]
    ) -> tp.Dict[str, tp.Any]:
        income_statement = IncomeStatements.process_financials_quote(financials_quote)
        return {"income_statement": income_statement}

symbol: str pydantic-field required

name: str pydantic-field required

asset_type: Literal['EQUITY', 'CURRENCY', 'CRYPTOCURRENCY', 'ETF', 'FUTURE', 'INDEX'] pydantic-field required

currency: str pydantic-field required

latest_price: float pydantic-field

sector: str pydantic-field

industry: str pydantic-field

earnings: Earnings pydantic-field required

trends: Trends pydantic-field required

income_statement: IncomeStatements pydantic-field

symbol_upper(symbol: str) -> str classmethod

Source code in pstock/asset.py

@validator("symbol")
def symbol_upper(cls, symbol: str) -> str:
    return symbol.upper()

process_quote(quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any] classmethod

Source code in pstock/asset.py

@classmethod
def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
    data = get_asset_data_from_quote(quote)
    earnings = Earnings.process_quote(quote)
    trends = Trends.process_quote(quote)
    return {**data, "earnings": earnings, "trends": trends}

process_financials_quote(financials_quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any] classmethod

Source code in pstock/asset.py

@classmethod
def process_financials_quote(
    cls, financials_quote: tp.Dict[str, tp.Any]
) -> tp.Dict[str, tp.Any]:
    income_statement = IncomeStatements.process_financials_quote(financials_quote)
    return {"income_statement": income_statement}

Assets (BaseModelSequence) pydantic-model

Source code in pstock/asset.py

class Assets(BaseModelSequence[Asset]):
    __root__: tp.List[Asset]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        df["earnings"] = df["earnings"].apply(lambda v: None if len(v) == 0 else v)
        df["trends"] = df["trends"].apply(lambda v: None if len(v) == 0 else v)
        return df.set_index("symbol").sort_index().dropna(axis=1, how="all")

    @classmethod
    async def get(
        cls,
        symbols: tp.List[str],
        *,
        client: tp.Optional[httpx.AsyncClient] = None,
    ):
        async with httpx_client_manager(client=client) as _client:
            async with asyncer.create_task_group() as tg:
                soon_values = [
                    tg.soonify(Asset.get)(symbol, client=_client) for symbol in symbols
                ]
        return cls.parse_obj([soon.value for soon in soon_values])

gen_df(self) -> pd.DataFrame

Source code in pstock/asset.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    df["earnings"] = df["earnings"].apply(lambda v: None if len(v) == 0 else v)
    df["trends"] = df["trends"].apply(lambda v: None if len(v) == 0 else v)
    return df.set_index("symbol").sort_index().dropna(axis=1, how="all")

get(symbols: tp.List[str], *, client: tp.Optional[httpx.AsyncClient] = None) async classmethod

Source code in pstock/asset.py

@classmethod
async def get(
    cls,
    symbols: tp.List[str],
    *,
    client: tp.Optional[httpx.AsyncClient] = None,
):
    async with httpx_client_manager(client=client) as _client:
        async with asyncer.create_task_group() as tg:
            soon_values = [
                tg.soonify(Asset.get)(symbol, client=_client) for symbol in symbols
            ]
    return cls.parse_obj([soon.value for soon in soon_values])

pstock.bar

EventParam

IntervalParam

PeriodParam

Bar (BaseModel) pydantic-model

Source code in pstock/bar.py

class Bar(BaseModel):
    date: datetime
    open: float
    high: float
    low: float
    close: float
    adj_close: float
    volume: float
    interval: timedelta

date: datetime pydantic-field required

open: float pydantic-field required

high: float pydantic-field required

low: float pydantic-field required

close: float pydantic-field required

adj_close: float pydantic-field required

volume: float pydantic-field required

interval: timedelta pydantic-field required

Bars (BaseModelSequence, _BarMixin) pydantic-model

Source code in pstock/bar.py

class Bars(BaseModelSequence[Bar], _BarMixin):
    __root__: tp.List[Bar]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        if not df.empty:
            df = df.dropna(
                how="all",
                subset=["open", "high", "low", "close", "adj_close", "volume"],
            )
            if df["interval"][0] >= timedelta(days=1):
                df["date"] = pd.to_datetime(pd.to_datetime(df["date"]).dt.date)
            df = df.set_index("date").sort_index()
        return df

    @classmethod
    def load(
        cls,
        *,
        response: tp.Union[ReadableResponse, str, bytes, dict],
    ) -> Bars:
        if isinstance(response, dict):
            data = response
        elif isinstance(response, (str, bytes)):
            data = json.loads(response)
        else:
            data = json.loads(response.read())

        return cls.parse_obj(get_ohlc_from_chart(data))

    @classmethod
    async def get(
        cls,
        symbol: str,
        *,
        interval: tp.Optional[IntervalParam] = None,
        period: tp.Optional[PeriodParam] = None,
        start: tp.Optional[Timestamp] = None,
        end: tp.Optional[Timestamp] = None,
        events: EventParam = "div,splits",
        include_prepost: bool = False,
        client: tp.Optional[httpx.AsyncClient] = None,
    ):
        url = cls.base_uri(symbol)
        params = cls.params(
            interval=interval,
            period=period,
            start=start,
            end=end,
            events=events,
            include_prepost=include_prepost,
        )
        async with httpx_client_manager(client=client) as _client:
            response = await _client.get(url, params=params)

        return cls.load(response=response)

gen_df(self) -> pd.DataFrame

Source code in pstock/bar.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    if not df.empty:
        df = df.dropna(
            how="all",
            subset=["open", "high", "low", "close", "adj_close", "volume"],
        )
        if df["interval"][0] >= timedelta(days=1):
            df["date"] = pd.to_datetime(pd.to_datetime(df["date"]).dt.date)
        df = df.set_index("date").sort_index()
    return df

load(*, response: tp.Union[ReadableResponse, str, bytes, dict]) -> Bars classmethod

Source code in pstock/bar.py

@classmethod
def load(
    cls,
    *,
    response: tp.Union[ReadableResponse, str, bytes, dict],
) -> Bars:
    if isinstance(response, dict):
        data = response
    elif isinstance(response, (str, bytes)):
        data = json.loads(response)
    else:
        data = json.loads(response.read())

    return cls.parse_obj(get_ohlc_from_chart(data))

get(symbol: str, *, interval: tp.Optional[IntervalParam] = None, period: tp.Optional[PeriodParam] = None, start: tp.Optional[Timestamp] = None, end: tp.Optional[Timestamp] = None, events: EventParam = 'div,splits', include_prepost: bool = False, client: tp.Optional[httpx.AsyncClient] = None) async classmethod

Source code in pstock/bar.py

@classmethod
async def get(
    cls,
    symbol: str,
    *,
    interval: tp.Optional[IntervalParam] = None,
    period: tp.Optional[PeriodParam] = None,
    start: tp.Optional[Timestamp] = None,
    end: tp.Optional[Timestamp] = None,
    events: EventParam = "div,splits",
    include_prepost: bool = False,
    client: tp.Optional[httpx.AsyncClient] = None,
):
    url = cls.base_uri(symbol)
    params = cls.params(
        interval=interval,
        period=period,
        start=start,
        end=end,
        events=events,
        include_prepost=include_prepost,
    )
    async with httpx_client_manager(client=client) as _client:
        response = await _client.get(url, params=params)

    return cls.load(response=response)

BarsMulti (BaseModelMapping, _BarMixin) pydantic-model

Source code in pstock/bar.py

class BarsMulti(BaseModelMapping[Bars], _BarMixin):
    __root__: tp.Dict[str, Bars]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        return df.sort_index()

    @classmethod
    async def get(
        cls,
        symbols: tp.List[str],
        *,
        interval: tp.Optional[IntervalParam] = None,
        period: tp.Optional[PeriodParam] = None,
        start: tp.Optional[Timestamp] = None,
        end: tp.Optional[Timestamp] = None,
        events: EventParam = "div,splits",
        include_prepost: bool = False,
        client: tp.Optional[httpx.AsyncClient] = None,
    ):
        async with httpx_client_manager(client=client) as _client:
            async with asyncer.create_task_group() as tg:
                soon_values = [
                    tg.soonify(Bars.get)(
                        symbol,
                        interval=interval,
                        period=period,
                        start=start,
                        end=end,
                        include_prepost=include_prepost,
                        events=events,
                        client=_client,
                    )
                    for symbol in symbols
                ]
        data = {
            symbol: soon_value.value for symbol, soon_value in zip(symbols, soon_values)
        }
        return cls.parse_obj(data)

gen_df(self) -> pd.DataFrame

Source code in pstock/bar.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    return df.sort_index()

get(symbols: tp.List[str], *, interval: tp.Optional[IntervalParam] = None, period: tp.Optional[PeriodParam] = None, start: tp.Optional[Timestamp] = None, end: tp.Optional[Timestamp] = None, events: EventParam = 'div,splits', include_prepost: bool = False, client: tp.Optional[httpx.AsyncClient] = None) async classmethod

Source code in pstock/bar.py

@classmethod
async def get(
    cls,
    symbols: tp.List[str],
    *,
    interval: tp.Optional[IntervalParam] = None,
    period: tp.Optional[PeriodParam] = None,
    start: tp.Optional[Timestamp] = None,
    end: tp.Optional[Timestamp] = None,
    events: EventParam = "div,splits",
    include_prepost: bool = False,
    client: tp.Optional[httpx.AsyncClient] = None,
):
    async with httpx_client_manager(client=client) as _client:
        async with asyncer.create_task_group() as tg:
            soon_values = [
                tg.soonify(Bars.get)(
                    symbol,
                    interval=interval,
                    period=period,
                    start=start,
                    end=end,
                    include_prepost=include_prepost,
                    events=events,
                    client=_client,
                )
                for symbol in symbols
            ]
    data = {
        symbol: soon_value.value for symbol, soon_value in zip(symbols, soon_values)
    }
    return cls.parse_obj(data)

pstock.base

T

U

BaseModel (BaseModel) pydantic-model

Source code in pstock/base.py

class BaseModel(_BaseModel):
    _created_at: datetime = PrivateAttr(default_factory=pendulum.now)

    @property
    def created_at(self) -> datetime:
        return self._created_at

created_at: datetime property readonly

BaseModelDf (BaseModel, ABC) pydantic-model

Source code in pstock/base.py

class BaseModelDf(BaseModel, ABC):

    _df: tp.Optional[pd.DataFrame] = PrivateAttr(default=None)

    @abstractmethod
    def gen_df(self) -> pd.DataFrame:
        ...

    @property
    def df(self) -> pd.DataFrame:
        if self._df is None:
            self._df = self.gen_df()
        return self._df

df: pd.DataFrame property readonly

gen_df(self) -> pd.DataFrame

Source code in pstock/base.py

@abstractmethod
def gen_df(self) -> pd.DataFrame:
    ...

BaseModelSequence (Generic, BaseModelDf) pydantic-model

Source code in pstock/base.py

class BaseModelSequence(tp.Generic[T], BaseModelDf):

    __root__: tp.Sequence[T]

    @tp.overload
    def __getitem__(self, index: int) -> T:
        """Get single item from __root__ by idx."""

    @tp.overload
    def __getitem__(self, index: slice) -> tp.Sequence[T]:
        """Get slice of items from __root__ by idx."""

    def __getitem__(self, index):
        return self.__root__[index]

    def __len__(self) -> int:
        return len(self.__root__)

    def __iter__(self) -> tp.Iterator[T]:  # type: ignore
        return iter(self.__root__)

    def gen_df(self) -> pd.DataFrame:
        return pd.DataFrame.from_dict(self.dict().get("__root__"), orient="columns")

__root__: Sequence[~T] pydantic-field required special

gen_df(self) -> pd.DataFrame

Source code in pstock/base.py

def gen_df(self) -> pd.DataFrame:
    return pd.DataFrame.from_dict(self.dict().get("__root__"), orient="columns")

BaseModelMapping (Generic, BaseModelDf) pydantic-model

Source code in pstock/base.py

class BaseModelMapping(tp.Generic[U], BaseModelDf):

    __root__: tp.Mapping[str, U]

    def __getitem__(self, index: str) -> U:
        return self.__root__[index]

    def __len__(self) -> int:
        return len(self.__root__)

    def __iter__(self) -> tp.Iterator[str]:  # type: ignore
        return iter(self.__root__)

    def gen_df(self) -> pd.DataFrame:
        keys, dfs = zip(*[(key, value.df) for key, value in self.__root__.items()])
        return pd.concat(dfs, axis=1, keys=keys)

__root__: Mapping[str, ~U] pydantic-field required special

gen_df(self) -> pd.DataFrame

Source code in pstock/base.py

def gen_df(self) -> pd.DataFrame:
    keys, dfs = zip(*[(key, value.df) for key, value in self.__root__.items()])
    return pd.concat(dfs, axis=1, keys=keys)

pstock.earnings

Earning (BaseModel) pydantic-model

Source code in pstock/earnings.py

class Earning(BaseModel):
    quarter: str
    estimate: float
    actual: float
    status: tp.Literal[None, "Beat", "Missed"] = None
    revenue: float
    earnings: float

    @validator("status", always=True)
    def set_status(
        cls, value: tp.Any, values: tp.Dict[str, tp.Any]
    ) -> tp.Literal[None, "Beat", "Missed"]:
        if value is not None:
            return value
        estimate = values.get("estimate")
        actual = values.get("actual")
        if actual is None or np.isnan(actual) or estimate is None or np.isnan(estimate):
            return None
        return "Beat" if actual >= estimate else "Missed"

quarter: str pydantic-field required

estimate: float pydantic-field required

actual: float pydantic-field required

status: Literal[None, 'Beat', 'Missed'] pydantic-field

revenue: float pydantic-field required

earnings: float pydantic-field required

set_status(value: Any, values: Dict[str, Any]) -> Literal[None, 'Beat', 'Missed'] classmethod

Source code in pstock/earnings.py

@validator("status", always=True)
def set_status(
    cls, value: tp.Any, values: tp.Dict[str, tp.Any]
) -> tp.Literal[None, "Beat", "Missed"]:
    if value is not None:
        return value
    estimate = values.get("estimate")
    actual = values.get("actual")
    if actual is None or np.isnan(actual) or estimate is None or np.isnan(estimate):
        return None
    return "Beat" if actual >= estimate else "Missed"

Earnings (BaseModelSequence, QuoteSummary) pydantic-model

Source code in pstock/earnings.py

class Earnings(BaseModelSequence[Earning], QuoteSummary):
    __root__: tp.List[Earning]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        if not df.empty:
            df = df.set_index("quarter").sort_index(key=pd.to_datetime)
        return df

    @validator("__root__")
    def sort_earnings(cls, value: tp.List[Earning]) -> tp.List[Earning]:
        if not value:
            return value
        return sorted(value, key=lambda earning: pd.to_datetime(earning.quarter))

    @classmethod
    def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
        return {"__root__": get_earnings_data_from_quote(quote)}

gen_df(self) -> DataFrame

Source code in pstock/earnings.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    if not df.empty:
        df = df.set_index("quarter").sort_index(key=pd.to_datetime)
    return df

sort_earnings(value: List[pstock.earnings.Earning]) -> List[pstock.earnings.Earning] classmethod

Source code in pstock/earnings.py

@validator("__root__")
def sort_earnings(cls, value: tp.List[Earning]) -> tp.List[Earning]:
    if not value:
        return value
    return sorted(value, key=lambda earning: pd.to_datetime(earning.quarter))

process_quote(quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/earnings.py

@classmethod
def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
    return {"__root__": get_earnings_data_from_quote(quote)}

pstock.income_statement

IncomeStatement (BaseModel) pydantic-model

Source code in pstock/income_statement.py

class IncomeStatement(BaseModel):
    date: date
    ebit: float
    total_revenue: float
    gross_profit: float

date: date pydantic-field required

ebit: float pydantic-field required

total_revenue: float pydantic-field required

gross_profit: float pydantic-field required

BaseIncomeStatements (BaseModelSequence, QuoteSummary) pydantic-model

Source code in pstock/income_statement.py

class BaseIncomeStatements(BaseModelSequence[IncomeStatement], QuoteSummary):
    __root__: tp.List[IncomeStatement]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        if not df.empty:
            df = df.set_index("date").sort_index()
        return df

gen_df(self) -> DataFrame

Source code in pstock/income_statement.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    if not df.empty:
        df = df.set_index("date").sort_index()
    return df

IncomeStatements (BaseIncomeStatements) pydantic-model

Source code in pstock/income_statement.py

class IncomeStatements(BaseIncomeStatements):
    @classmethod
    def process_financials_quote(
        cls, financials_quote: tp.Dict[str, tp.Any]
    ) -> tp.Dict[str, tp.Any]:
        return {
            "__root__": get_income_statement_data_from_financials_quote(
                financials_quote, key="incomeStatementHistory"
            )
        }

    @property
    def revenue_compound_annual_growth_rate(self) -> float:
        """Revenue CAGR.

        https://www.investopedia.com/terms/c/cagr.asp

        Returns:
            float: Revenue CAGR based
        """
        latest_revenue = self.df.max()["total_revenue"]
        earliest_revenue = self.df.min()["total_revenue"]
        num_years = len(self.df) - 1
        return (latest_revenue / earliest_revenue) ** (1 / num_years) - 1

revenue_compound_annual_growth_rate: float property readonly

Revenue CAGR.

https://www.investopedia.com/terms/c/cagr.asp

Returns:

Type	Description
float	Revenue CAGR based

process_financials_quote(financials_quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/income_statement.py

@classmethod
def process_financials_quote(
    cls, financials_quote: tp.Dict[str, tp.Any]
) -> tp.Dict[str, tp.Any]:
    return {
        "__root__": get_income_statement_data_from_financials_quote(
            financials_quote, key="incomeStatementHistory"
        )
    }

QuarterlyIncomeStatements (BaseIncomeStatements) pydantic-model

Source code in pstock/income_statement.py

class QuarterlyIncomeStatements(BaseIncomeStatements):
    @classmethod
    def process_financials_quote(
        cls, financials_quote: tp.Dict[str, tp.Any]
    ) -> tp.Dict[str, tp.Any]:
        return {
            "__root__": get_income_statement_data_from_financials_quote(
                financials_quote, key="incomeStatementHistoryQuarterly"
            )
        }

    @property
    def revenue_compound_annual_growth_rate(self) -> float:
        raise NotImplementedError()

revenue_compound_annual_growth_rate: float property readonly

process_financials_quote(financials_quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/income_statement.py

@classmethod
def process_financials_quote(
    cls, financials_quote: tp.Dict[str, tp.Any]
) -> tp.Dict[str, tp.Any]:
    return {
        "__root__": get_income_statement_data_from_financials_quote(
            financials_quote, key="incomeStatementHistoryQuarterly"
        )
    }

pstock.news

Publication (BaseModel) pydantic-model

Source code in pstock/news.py

class Publication(BaseModel):
    date: datetime.datetime
    title: str
    url: str
    summary: tp.Optional[str]

date: datetime pydantic-field required

title: str pydantic-field required

url: str pydantic-field required

summary: str pydantic-field

News (BaseModelSequence) pydantic-model

Source code in pstock/news.py

class News(BaseModelSequence):
    __root__: tp.List[Publication]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        if not df.empty:
            df = df.set_index("date").sort_index()
        return df

    @staticmethod
    def base_uri() -> str:
        return "https://feeds.finance.yahoo.com/rss/2.0/headline"

    @staticmethod
    def params(symbol: str) -> tp.Dict[str, str]:
        return {"s": symbol.upper(), "region": "US", "lang": "en-US"}

    @classmethod
    def uri(cls, symbol: str) -> str:
        return f"{cls.base_uri()}?{urlencode(cls.params(symbol))}"

    @classmethod
    def load(
        cls,
        *,
        symbol: tp.Optional[str] = None,
        response: tp.Union[None, str, bytes, ReadableResponse] = None,
    ) -> News:
        if symbol is not None and response is None:
            # feedparser can take a uri as input, and get data over http
            response = cls.uri(symbol)
            print(response)

        if response is None:
            raise ValueError(
                "Please provide either a symbol or or a readeable response."
            )

        feed = feedparser.parse(response)
        return cls.parse_obj(
            [
                {
                    **entry,
                    "date": time.mktime(entry["published_parsed"]),
                    "url": entry["link"],
                }
                for entry in feed.entries
            ]
        )

    @classmethod
    async def get(
        cls,
        symbol: str,
        *,
        client: tp.Optional[httpx.AsyncClient] = None,
    ) -> News:
        async with httpx_client_manager(client=client) as _client:
            response = await _client.get(cls.base_uri(), params=cls.params(symbol))

        return cls.load(response=response)

gen_df(self) -> pd.DataFrame

Source code in pstock/news.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    if not df.empty:
        df = df.set_index("date").sort_index()
    return df

base_uri() -> str staticmethod

Source code in pstock/news.py

@staticmethod
def base_uri() -> str:
    return "https://feeds.finance.yahoo.com/rss/2.0/headline"

params(symbol: str) -> tp.Dict[str, str] staticmethod

Source code in pstock/news.py

@staticmethod
def params(symbol: str) -> tp.Dict[str, str]:
    return {"s": symbol.upper(), "region": "US", "lang": "en-US"}

uri(symbol: str) -> str classmethod

Source code in pstock/news.py

@classmethod
def uri(cls, symbol: str) -> str:
    return f"{cls.base_uri()}?{urlencode(cls.params(symbol))}"

load(*, symbol: tp.Optional[str] = None, response: tp.Union[None, str, bytes, ReadableResponse] = None) -> News classmethod

Source code in pstock/news.py

@classmethod
def load(
    cls,
    *,
    symbol: tp.Optional[str] = None,
    response: tp.Union[None, str, bytes, ReadableResponse] = None,
) -> News:
    if symbol is not None and response is None:
        # feedparser can take a uri as input, and get data over http
        response = cls.uri(symbol)
        print(response)

    if response is None:
        raise ValueError(
            "Please provide either a symbol or or a readeable response."
        )

    feed = feedparser.parse(response)
    return cls.parse_obj(
        [
            {
                **entry,
                "date": time.mktime(entry["published_parsed"]),
                "url": entry["link"],
            }
            for entry in feed.entries
        ]
    )

get(symbol: str, *, client: tp.Optional[httpx.AsyncClient] = None) -> News async classmethod

Source code in pstock/news.py

@classmethod
async def get(
    cls,
    symbol: str,
    *,
    client: tp.Optional[httpx.AsyncClient] = None,
) -> News:
    async with httpx_client_manager(client=client) as _client:
        response = await _client.get(cls.base_uri(), params=cls.params(symbol))

    return cls.load(response=response)

pstock.quote

T

QuoteSummary (BaseModel) pydantic-model

Source code in pstock/quote.py

class QuoteSummary(BaseModel):
    @staticmethod
    def uri(symbol: str) -> str:
        return f"https://finance.yahoo.com/quote/{symbol.upper()}"

    @staticmethod
    def financials_uri(symbol: str) -> str:
        return (
            f"https://finance.yahoo.com/quote/{symbol.upper()}/"
            f"financials?p={symbol.upper()}"
        )

    @staticmethod
    def parse_quote(
        response: tp.Union[ReadableResponse, str, bytes]
    ) -> tp.Dict[str, tp.Any]:

        content = response if isinstance(response, (str, bytes)) else response.read()

        soup = BeautifulSoup(content, "html.parser")

        script = soup.find("script", text=re.compile(r"root.App.main"))
        if script is None:
            return {}
        match = re.search(r"root.App.main\s+=\s+(\{.*\})", script.text)

        if match is None:
            return {}

        data: tp.Dict[str, tp.Any] = json.loads(match.group(1))
        return (
            data.get("context", {})
            .get("dispatcher", {})
            .get("stores", {})
            .get("QuoteSummaryStore", {})
        )

    @classmethod
    def process_quote(
        cls: tp.Type[T], quote: tp.Dict[str, tp.Any]
    ) -> tp.Dict[str, tp.Any]:
        return {}

    @classmethod
    def process_financials_quote(
        cls: tp.Type[T], financials_quote: tp.Dict[str, tp.Any]
    ) -> tp.Dict[str, tp.Any]:
        return {}

    @classmethod
    def load(
        cls: tp.Type[T],
        *,
        response: tp.Union[ReadableResponse, str, bytes, None] = None,
        financials_response: tp.Union[ReadableResponse, str, bytes, None] = None,
    ) -> T:

        data = {}
        _quote = None
        _financials_quote = None

        if response is not None:
            _quote = cls.parse_quote(response)
            if _quote:
                data.update(cls.process_quote(_quote))

        if financials_response is not None:
            _financials_quote = cls.parse_quote(financials_response)
            if _financials_quote:
                data.update(cls.process_financials_quote(_financials_quote))

        return cls(**data)

    @classmethod
    async def get(
        cls: tp.Type[T],
        symbol: str,
        *,
        client: tp.Optional[httpx.AsyncClient] = None,
    ) -> T:
        async with httpx_client_manager(client=client) as _client:
            async with asyncer.create_task_group() as tg:
                soon_quote = tg.soonify(_client.get)(
                    cls.uri(symbol), headers={"user-agent": rdm_user_agent_value()}
                )
                soon_financials = tg.soonify(_client.get)(
                    cls.financials_uri(symbol),
                    headers={"user-agent": rdm_user_agent_value()},
                )

        return cls.load(
            response=soon_quote.value, financials_response=soon_financials.value
        )

uri(symbol: str) -> str staticmethod

Source code in pstock/quote.py

@staticmethod
def uri(symbol: str) -> str:
    return f"https://finance.yahoo.com/quote/{symbol.upper()}"

financials_uri(symbol: str) -> str staticmethod

Source code in pstock/quote.py

@staticmethod
def financials_uri(symbol: str) -> str:
    return (
        f"https://finance.yahoo.com/quote/{symbol.upper()}/"
        f"financials?p={symbol.upper()}"
    )

parse_quote(response: Union[pstock.types.ReadableResponse, str, bytes]) -> Dict[str, Any] staticmethod

Source code in pstock/quote.py

@staticmethod
def parse_quote(
    response: tp.Union[ReadableResponse, str, bytes]
) -> tp.Dict[str, tp.Any]:

    content = response if isinstance(response, (str, bytes)) else response.read()

    soup = BeautifulSoup(content, "html.parser")

    script = soup.find("script", text=re.compile(r"root.App.main"))
    if script is None:
        return {}
    match = re.search(r"root.App.main\s+=\s+(\{.*\})", script.text)

    if match is None:
        return {}

    data: tp.Dict[str, tp.Any] = json.loads(match.group(1))
    return (
        data.get("context", {})
        .get("dispatcher", {})
        .get("stores", {})
        .get("QuoteSummaryStore", {})
    )

process_quote(quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/quote.py

@classmethod
def process_quote(
    cls: tp.Type[T], quote: tp.Dict[str, tp.Any]
) -> tp.Dict[str, tp.Any]:
    return {}

process_financials_quote(financials_quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/quote.py

@classmethod
def process_financials_quote(
    cls: tp.Type[T], financials_quote: tp.Dict[str, tp.Any]
) -> tp.Dict[str, tp.Any]:
    return {}

load(*, response: Union[pstock.types.ReadableResponse, str, bytes] = None, financials_response: Union[pstock.types.ReadableResponse, str, bytes] = None) -> ~T classmethod

Source code in pstock/quote.py

@classmethod
def load(
    cls: tp.Type[T],
    *,
    response: tp.Union[ReadableResponse, str, bytes, None] = None,
    financials_response: tp.Union[ReadableResponse, str, bytes, None] = None,
) -> T:

    data = {}
    _quote = None
    _financials_quote = None

    if response is not None:
        _quote = cls.parse_quote(response)
        if _quote:
            data.update(cls.process_quote(_quote))

    if financials_response is not None:
        _financials_quote = cls.parse_quote(financials_response)
        if _financials_quote:
            data.update(cls.process_financials_quote(_financials_quote))

    return cls(**data)

get(symbol: str, *, client: Optional[httpx.AsyncClient] = None) -> ~T async classmethod

Source code in pstock/quote.py

@classmethod
async def get(
    cls: tp.Type[T],
    symbol: str,
    *,
    client: tp.Optional[httpx.AsyncClient] = None,
) -> T:
    async with httpx_client_manager(client=client) as _client:
        async with asyncer.create_task_group() as tg:
            soon_quote = tg.soonify(_client.get)(
                cls.uri(symbol), headers={"user-agent": rdm_user_agent_value()}
            )
            soon_financials = tg.soonify(_client.get)(
                cls.financials_uri(symbol),
                headers={"user-agent": rdm_user_agent_value()},
            )

    return cls.load(
        response=soon_quote.value, financials_response=soon_financials.value
    )

pstock.trend

Trend (BaseModel) pydantic-model

Source code in pstock/trend.py

class Trend(BaseModel):
    date: datetime.date
    strong_buy: int = 0
    buy: int = 0
    hold: int = 0
    sell: int = 0
    strong_sell: int = 0
    score: tp.Optional[float] = None
    recomendation: tp.Literal[
        None, "STRONG_BUY", "BUY", "HOLD", "SELL", "STRONG_SELL"
    ] = None

    @validator("score", always=True)
    def compute_score(
        cls, value: tp.Optional[float], values: tp.Dict[str, tp.Any]
    ) -> float:
        if value is not None:
            return value
        numerator = (
            values["strong_buy"]
            + values["buy"] * 2
            + values["hold"] * 3
            + values["sell"] * 4
            + values["strong_sell"] * 5
        )
        denominator = (
            values["strong_buy"]
            + values["buy"]
            + values["hold"]
            + values["sell"]
            + values["strong_sell"]
        )
        if denominator == 0:
            return np.nan
        return round(numerator / denominator, 2)

    @validator("recomendation", always=True)
    def compute_recomendation(
        cls, value: tp.Optional[float], values: tp.Dict[str, tp.Any]
    ):
        if value is not None:
            return value
        score = values["score"]
        if np.isnan(score):
            return "UNKNOWN"
        elif score >= 4.5:
            return "STRONG_SELL"
        elif score >= 3.5:
            return "SELL"
        elif score >= 2.5:
            return "HOLD"
        elif score >= 1.5:
            return "BUY"
        else:
            return "STRONG_BUY"

date: date pydantic-field required

strong_buy: int pydantic-field

buy: int pydantic-field

hold: int pydantic-field

sell: int pydantic-field

strong_sell: int pydantic-field

score: float pydantic-field

recomendation: Literal[None, 'STRONG_BUY', 'BUY', 'HOLD', 'SELL', 'STRONG_SELL'] pydantic-field

compute_score(value: Optional[float], values: Dict[str, Any]) -> float classmethod

Source code in pstock/trend.py

@validator("score", always=True)
def compute_score(
    cls, value: tp.Optional[float], values: tp.Dict[str, tp.Any]
) -> float:
    if value is not None:
        return value
    numerator = (
        values["strong_buy"]
        + values["buy"] * 2
        + values["hold"] * 3
        + values["sell"] * 4
        + values["strong_sell"] * 5
    )
    denominator = (
        values["strong_buy"]
        + values["buy"]
        + values["hold"]
        + values["sell"]
        + values["strong_sell"]
    )
    if denominator == 0:
        return np.nan
    return round(numerator / denominator, 2)

compute_recomendation(value: Optional[float], values: Dict[str, Any]) classmethod

Source code in pstock/trend.py

@validator("recomendation", always=True)
def compute_recomendation(
    cls, value: tp.Optional[float], values: tp.Dict[str, tp.Any]
):
    if value is not None:
        return value
    score = values["score"]
    if np.isnan(score):
        return "UNKNOWN"
    elif score >= 4.5:
        return "STRONG_SELL"
    elif score >= 3.5:
        return "SELL"
    elif score >= 2.5:
        return "HOLD"
    elif score >= 1.5:
        return "BUY"
    else:
        return "STRONG_BUY"

Trends (BaseModelSequence, QuoteSummary) pydantic-model

Source code in pstock/trend.py

class Trends(BaseModelSequence[Trend], QuoteSummary):
    __root__: tp.List[Trend]

    def gen_df(self) -> pd.DataFrame:
        df = super().gen_df()
        if not df.empty:
            df = df.set_index("date").sort_index()
        return df

    @validator("__root__")
    def sort_trends(cls, value: tp.List[Trend]) -> tp.List[Trend]:
        if not value:
            return value
        return sorted(value, key=lambda trend: pd.to_datetime(trend.date))

    @classmethod
    def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
        return {"__root__": get_trends_data_from_quote(quote)}

gen_df(self) -> DataFrame

Source code in pstock/trend.py

def gen_df(self) -> pd.DataFrame:
    df = super().gen_df()
    if not df.empty:
        df = df.set_index("date").sort_index()
    return df

sort_trends(value: List[pstock.trend.Trend]) -> List[pstock.trend.Trend] classmethod

Source code in pstock/trend.py

@validator("__root__")
def sort_trends(cls, value: tp.List[Trend]) -> tp.List[Trend]:
    if not value:
        return value
    return sorted(value, key=lambda trend: pd.to_datetime(trend.date))

process_quote(quote: Dict[str, Any]) -> Dict[str, Any] classmethod

Source code in pstock/trend.py

@classmethod
def process_quote(cls, quote: tp.Dict[str, tp.Any]) -> tp.Dict[str, tp.Any]:
    return {"__root__": get_trends_data_from_quote(quote)}

pstock.types

Timestamp (int)

Source code in pstock/types.py

class Timestamp(int):
    @classmethod
    def __get_validators__(cls):
        yield cls.validate

    @classmethod
    def validate(cls, value: tp.Union[str, int, float, date, datetime]) -> int:
        return parse_datetime(value).int_timestamp

validate(value: Union[str, int, float, datetime.date, datetime.datetime]) -> int classmethod

Source code in pstock/types.py

@classmethod
def validate(cls, value: tp.Union[str, int, float, date, datetime]) -> int:
    return parse_datetime(value).int_timestamp

ReadableResponse (Protocol)

Source code in pstock/types.py

class ReadableResponse(tp.Protocol):
    def read(self) -> tp.Union[str, bytes]:
        ...

read(self) -> Union[str, bytes]

Source code in pstock/types.py

def read(self) -> tp.Union[str, bytes]:
    ...

__init__(self, *args, **kwargs) special

Source code in pstock/types.py

def _no_init_or_replace_init(self, *args, **kwargs):
    cls = type(self)

    if cls._is_protocol:
        raise TypeError('Protocols cannot be instantiated')

    # Already using a custom `__init__`. No need to calculate correct
    # `__init__` to call. This can lead to RecursionError. See bpo-45121.
    if cls.__init__ is not _no_init_or_replace_init:
        return

    # Initially, `__init__` of a protocol subclass is set to `_no_init_or_replace_init`.
    # The first instantiation of the subclass will call `_no_init_or_replace_init` which
    # searches for a proper new `__init__` in the MRO. The new `__init__`
    # replaces the subclass' old `__init__` (ie `_no_init_or_replace_init`). Subsequent
    # instantiation of the protocol subclass will thus use the new
    # `__init__` and no longer call `_no_init_or_replace_init`.
    for base in cls.__mro__:
        init = base.__dict__.get('__init__', _no_init_or_replace_init)
        if init is not _no_init_or_replace_init:
            cls.__init__ = init
            break
    else:
        # should not happen
        cls.__init__ = object.__init__

    cls.__init__(self, *args, **kwargs)