from __future__ import annotations
import typing
from . import models
from ._runtime import AsyncPager, AsyncTransport, SyncPager, Transport, path_seg, query_of, to_wire
from ._server import EmbedConfig, embedded_base_url
DEFAULT_BASE_URL = "https://finance-query.com"
EMBED = EmbedConfig(
binary="finance-query-server",
args=(),
base_url_env="FINANCE_QUERY_BASE_URL",
bin_env="FINANCE_QUERY_SERVER_BIN",
env=(
("HOST", "127.0.0.1"),
("PORT", "0"),
),
)
class ServerEnv(typing.TypedDict, total=False):
PORT: str
HOST: str
LOG_LEVEL: str
LOG_FORMAT: str
REDIS_URL: str
RATE_LIMIT_PER_MINUTE: str
DEFAULT_INTERVAL: str
DEFAULT_RANGE: str
RECOMMENDATIONS_LIMIT: str
SEARCH_HITS: str
SCREENERS_COUNT: str
EDGAR_EMAIL: str
FRED_API_KEY: str
POLYGON_API_KEY: str
FMP_API_KEY: str
ALPHAVANTAGE_API_KEY: str
BLS_API_KEY: str
OPENFIGI_API_KEY: str
class Client:
def __init__(
self,
base_url: str | None = None,
*,
server_env: ServerEnv | typing.Mapping[str, str] | None = None,
timeout: float = 30.0,
max_retries: int = 3,
on_rate_limit: typing.Callable[[typing.Any], None] | None = None,
transport: Transport | None = None,
) -> None:
base_url = base_url or embedded_base_url(EMBED, typing.cast("typing.Mapping[str, str] | None", server_env))
self._transport = transport or Transport(
base_url, timeout=timeout, max_retries=max_retries, on_rate_limit=on_rate_limit
)
def close(self) -> None:
self._transport.close()
def __enter__(self) -> Client:
return self
def __exit__(self, *exc: object) -> None:
self.close()
def get_analysis(
self,
symbol: str,
type: models.AnalysisType,
*,
fields: str | None = None,
) -> models.AnalysisResponse:
return self._transport.request(
"GET",
f"/v2/analysis/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields}),
into=models.AnalysisResponse,
)
def get_batch_capital_gains(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlCapitalGainsBatch:
return self._transport.request(
"GET",
"/v2/capital-gains",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlCapitalGainsBatch,
)
def get_batch_charts(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlChartsBatch:
return self._transport.request(
"GET",
"/v2/charts",
query=query_of({"fields": fields, "interval": interval, "patterns": patterns, "range": range, "symbols": symbols}),
into=models.GqlChartsBatch,
)
def iter_batch_charts(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/charts",
query=query_of({"fields": fields, "interval": interval, "patterns": patterns, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_batch_dividends(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlDividendsBatch:
return self._transport.request(
"GET",
"/v2/dividends",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlDividendsBatch,
)
def iter_batch_dividends(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/dividends",
query=query_of({"fields": fields, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_batch_financials(
self,
*,
fields: str | None = None,
frequency: models.Frequency | None = None,
metrics: str | None = None,
statement: models.StatementType,
symbols: str,
) -> models.GqlFinancialsBatch:
return self._transport.request(
"GET",
"/v2/financials",
query=query_of({"fields": fields, "frequency": frequency, "metrics": metrics, "statement": statement, "symbols": symbols}),
into=models.GqlFinancialsBatch,
)
def get_batch_indicators(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlIndicatorsBatch:
return self._transport.request(
"GET",
"/v2/indicators",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols}),
into=models.GqlIndicatorsBatch,
)
def iter_batch_indicators(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/indicators",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_batch_options(
self,
*,
date: int | None = None,
fields: str | None = None,
symbols: str,
) -> models.GqlOptionsBatch:
return self._transport.request(
"GET",
"/v2/options",
query=query_of({"date": date, "fields": fields, "symbols": symbols}),
into=models.GqlOptionsBatch,
)
def iter_batch_options(
self,
*,
date: int | None = None,
fields: str | None = None,
symbols: str,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/options",
query=query_of({"date": date, "fields": fields, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_batch_recommendations(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
symbols: str,
) -> models.GqlRecommendationsBatch:
return self._transport.request(
"GET",
"/v2/recommendations",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit, "symbols": symbols}),
into=models.GqlRecommendationsBatch,
)
def get_batch_splits(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlSplitsBatch:
return self._transport.request(
"GET",
"/v2/splits",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlSplitsBatch,
)
def get_calendar(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> list[models.GqlCalendarEvent]:
return self._transport.request(
"GET",
"/v2/calendar",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=list[models.GqlCalendarEvent],
)
def get_capital_gains(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> list[models.GqlCapitalGain]:
return self._transport.request(
"GET",
f"/v2/capital-gains/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=list[models.GqlCapitalGain],
)
def get_chart(
self,
symbol: str,
*,
end: int | None = None,
events: bool | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
start: int | None = None,
) -> models.GqlChart:
return self._transport.request(
"GET",
f"/v2/chart/{path_seg(symbol)}",
query=query_of({"end": end, "events": events, "fields": fields, "interval": interval, "patterns": patterns, "range": range, "start": start}),
into=models.GqlChart,
)
def iter_chart(
self,
symbol: str,
*,
end: int | None = None,
events: bool | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
start: int | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/chart/{path_seg(symbol)}",
query=query_of({"end": end, "events": events, "fields": fields, "interval": interval, "patterns": patterns, "range": range, "start": start, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_commitments_of_traders(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> models.GqlCommitmentsOfTraders:
return self._transport.request(
"GET",
f"/v2/cftc/cot/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=models.GqlCommitmentsOfTraders,
)
def get_commodity(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlCommodityQuote:
return self._transport.request(
"GET",
f"/v2/commodities/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlCommodityQuote,
)
def get_company_profile(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlCompanyProfile:
return self._transport.request(
"GET",
f"/v2/company-profile/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlCompanyProfile,
)
def get_congressional_trades(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlCongressionalTrade]:
return self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/congressional-trades",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlCongressionalTrade],
)
def get_crypto_coin(
self,
id: str,
*,
fields: str | None = None,
vs_currency: str | None = None,
) -> models.GqlCoinQuote:
return self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}",
query=query_of({"fields": fields, "vsCurrency": vs_currency}),
into=models.GqlCoinQuote,
)
def get_crypto_coins(
self,
*,
count: int | None = None,
fields: str | None = None,
vs_currency: str | None = None,
) -> list[models.GqlCoinQuote]:
return self._transport.request(
"GET",
"/v2/crypto/coins",
query=query_of({"count": count, "fields": fields, "vsCurrency": vs_currency}),
into=list[models.GqlCoinQuote],
)
def iter_crypto_coins(
self,
*,
count: int | None = None,
fields: str | None = None,
vs_currency: str | None = None,
limit: int = 50,
) -> SyncPager[models.GqlCoinQuote]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/crypto/coins",
query=query_of({"count": count, "fields": fields, "vsCurrency": vs_currency, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlCoinQuote)
def get_crypto_global(
self,
*,
fields: str | None = None,
) -> models.GqlGlobalCryptoStats:
return self._transport.request(
"GET",
"/v2/crypto/global",
query=query_of({"fields": fields}),
into=models.GqlGlobalCryptoStats,
)
def get_crypto_news(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
) -> list[models.GqlNews]:
return self._transport.request(
"GET",
"/v2/crypto/news",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit}),
into=list[models.GqlNews],
)
def get_crypto_search(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
limit_results: int | None = None,
query: str,
) -> list[models.GqlSymbolMatch]:
return self._transport.request(
"GET",
"/v2/crypto/search",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit, "limitResults": limit_results, "query": query}),
into=list[models.GqlSymbolMatch],
)
def get_crypto_trending(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlTrendingCoin]:
return self._transport.request(
"GET",
"/v2/crypto/trending",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlTrendingCoin],
)
def get_currencies(
self,
*,
fields: str | None = None,
) -> list[models.GqlCurrency]:
return self._transport.request(
"GET",
"/v2/currencies",
query=query_of({"fields": fields}),
into=list[models.GqlCurrency],
)
def get_dividends(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> models.GqlDividends:
return self._transport.request(
"GET",
f"/v2/dividends/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=models.GqlDividends,
)
def iter_dividends(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/dividends/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_earnings_surprises(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEarningsSurpriseHistory:
return self._transport.request(
"GET",
f"/v2/earnings-surprises/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEarningsSurpriseHistory,
)
def get_earnings_transcript(
self,
symbol: str,
*,
fields: str | None = None,
quarter: models.Quarter | None = None,
year: int | None = None,
) -> models.GqlEarningsTranscript:
return self._transport.request(
"GET",
f"/v2/earnings-transcript/{path_seg(symbol)}",
query=query_of({"fields": fields, "quarter": quarter, "year": year}),
into=models.GqlEarningsTranscript,
)
def get_edgar_cik(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEdgarCik:
return self._transport.request(
"GET",
f"/v2/edgar/cik/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEdgarCik,
)
def get_edgar_facts(
self,
symbol: str,
*,
concepts: str | None = None,
fields: str | None = None,
taxonomy: str | None = None,
) -> list[models.GqlFactConcept]:
return self._transport.request(
"GET",
f"/v2/edgar/facts/{path_seg(symbol)}",
query=query_of({"concepts": concepts, "fields": fields, "taxonomy": taxonomy}),
into=list[models.GqlFactConcept],
)
def iter_edgar_facts(
self,
symbol: str,
*,
concepts: str | None = None,
fields: str | None = None,
taxonomy: str | None = None,
limit: int = 50,
) -> SyncPager[models.GqlFactConcept]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/edgar/facts/{path_seg(symbol)}",
query=query_of({"concepts": concepts, "fields": fields, "taxonomy": taxonomy, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlFactConcept)
def get_edgar_search(
self,
*,
end_date: str | None = None,
fields: str | None = None,
forms: str | None = None,
from_: int | None = None,
q: str,
size: int | None = None,
start_date: str | None = None,
) -> models.GqlEdgarSearchResults:
return self._transport.request(
"GET",
"/v2/edgar/search",
query=query_of({"endDate": end_date, "fields": fields, "forms": forms, "from": from_, "q": q, "size": size, "startDate": start_date}),
into=models.GqlEdgarSearchResults,
)
def get_edgar_submissions(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEdgarSubmissions:
return self._transport.request(
"GET",
f"/v2/edgar/submissions/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEdgarSubmissions,
)
def iter_edgar_submissions(
self,
symbol: str,
*,
fields: str | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/edgar/submissions/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_etf_profile(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEtfProfile:
return self._transport.request(
"GET",
f"/v2/etf-profile/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEtfProfile,
)
def get_exchanges(
self,
*,
fields: str | None = None,
) -> list[models.GqlExchange]:
return self._transport.request(
"GET",
"/v2/exchanges",
query=query_of({"fields": fields}),
into=list[models.GqlExchange],
)
def get_executive_compensation(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlExecutiveCompensation]:
return self._transport.request(
"GET",
f"/v2/executive-compensation/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlExecutiveCompensation],
)
def get_fails_to_deliver(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlFailToDeliver]:
return self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/fails-to-deliver",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlFailToDeliver],
)
def get_fear_and_greed(
self,
*,
fields: str | None = None,
) -> models.GqlFearAndGreed:
return self._transport.request(
"GET",
"/v2/fear-and-greed",
query=query_of({"fields": fields}),
into=models.GqlFearAndGreed,
)
def get_feeds(
self,
*,
fields: str | None = None,
form_type: str | None = None,
sources: str | None = None,
) -> list[models.GqlFeedEntry]:
return self._transport.request(
"GET",
"/v2/feeds",
query=query_of({"fields": fields, "formType": form_type, "sources": sources}),
into=list[models.GqlFeedEntry],
)
def iter_feeds(
self,
*,
fields: str | None = None,
form_type: str | None = None,
sources: str | None = None,
limit: int = 50,
) -> SyncPager[models.GqlFeedEntry]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/feeds",
query=query_of({"fields": fields, "formType": form_type, "sources": sources, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlFeedEntry)
def get_filing_sections(
self,
symbol: str,
*,
accession_number: str,
fields: str | None = None,
form: models.FilingSectionFormParam,
) -> list[models.GqlFilingSection]:
return self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/sections",
query=query_of({"accessionNumber": accession_number, "fields": fields, "form": form}),
into=list[models.GqlFilingSection],
)
def get_financials(
self,
symbol: str,
statement: models.StatementType,
*,
fields: str | None = None,
frequency: models.Frequency | None = None,
metrics: str | None = None,
) -> list[models.GqlFinancialLineItem]:
return self._transport.request(
"GET",
f"/v2/financials/{path_seg(symbol)}/{path_seg(statement)}",
query=query_of({"fields": fields, "frequency": frequency, "metrics": metrics}),
into=list[models.GqlFinancialLineItem],
)
def get_forex(
self,
from_: str,
to: str,
*,
fields: str | None = None,
) -> models.GqlForexQuote:
return self._transport.request(
"GET",
f"/v2/forex/{path_seg(from_)}/{path_seg(to)}",
query=query_of({"fields": fields}),
into=models.GqlForexQuote,
)
def get_forex_news(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
) -> list[models.GqlNews]:
return self._transport.request(
"GET",
"/v2/forex/news",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit}),
into=list[models.GqlNews],
)
def get_fred_series(
self,
id: str,
*,
fields: str | None = None,
) -> models.GqlMacroSeries:
return self._transport.request(
"GET",
f"/v2/fred/series/{path_seg(id)}",
query=query_of({"fields": fields}),
into=models.GqlMacroSeries,
)
def iter_fred_series(
self,
id: str,
*,
fields: str | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/fred/series/{path_seg(id)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_fred_treasury_yields(
self,
*,
fields: str | None = None,
year: int | None = None,
) -> list[models.GqlTreasuryYield]:
return self._transport.request(
"GET",
"/v2/fred/treasury-yields",
query=query_of({"fields": fields, "year": year}),
into=list[models.GqlTreasuryYield],
)
def iter_fred_treasury_yields(
self,
*,
fields: str | None = None,
year: int | None = None,
limit: int = 50,
) -> SyncPager[models.GqlTreasuryYield]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/fred/treasury-yields",
query=query_of({"fields": fields, "year": year, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlTreasuryYield)
def get_futures(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlFuturesQuote:
return self._transport.request(
"GET",
f"/v2/futures/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlFuturesQuote,
)
def get_gdelt_news(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlNews]:
return self._transport.request(
"GET",
f"/v2/gdelt/news/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlNews],
)
def get_general_news(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
) -> list[models.GqlNews]:
return self._transport.request(
"GET",
"/v2/news",
query=query_of({"count": count, "fields": fields, "lang": lang}),
into=list[models.GqlNews],
)
def iter_general_news(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
limit: int = 50,
) -> SyncPager[models.GqlNews]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/news",
query=query_of({"count": count, "fields": fields, "lang": lang, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlNews)
def get_grading_actions(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlGradingAction]:
return self._transport.request(
"GET",
f"/v2/grading-actions/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlGradingAction],
)
def get_holders(
self,
symbol: str,
type: models.HolderType,
*,
fields: str | None = None,
) -> models.HoldersResponse:
return self._transport.request(
"GET",
f"/v2/holders/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields}),
into=models.HoldersResponse,
)
def iter_holders(
self,
symbol: str,
type: models.HolderType,
*,
fields: str | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/holders/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_hours(
self,
*,
fields: str | None = None,
region: models.Region | None = None,
) -> models.GqlMarketHours:
return self._transport.request(
"GET",
"/v2/hours",
query=query_of({"fields": fields, "region": region}),
into=models.GqlMarketHours,
)
def get_index_constituent_changes(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlIndexConstituentChange]:
return self._transport.request(
"GET",
f"/v2/indices/{path_seg(symbol)}/constituent-changes",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlIndexConstituentChange],
)
def get_index_constituents(
self,
symbol: str,
*,
fields: str | None = None,
) -> list[models.GqlIndexConstituent]:
return self._transport.request(
"GET",
f"/v2/index-constituents/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=list[models.GqlIndexConstituent],
)
def get_indicators(
self,
symbol: str,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
) -> models.GqlIndicatorsSummary:
return self._transport.request(
"GET",
f"/v2/indicators/{path_seg(symbol)}",
query=query_of({"fields": fields, "interval": interval, "range": range}),
into=models.GqlIndicatorsSummary,
)
def get_indices(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
region: models.indices_Region | None = None,
) -> list[models.GqlQuote]:
return self._transport.request(
"GET",
"/v2/indices",
query=query_of({"fields": fields, "format": format, "region": region}),
into=list[models.GqlQuote],
)
def get_industry(
self,
industry: models.Industry,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
) -> models.GqlIndustryData:
return self._transport.request(
"GET",
f"/v2/industries/{path_seg(industry)}",
query=query_of({"fields": fields, "format": format, "lang": lang}),
into=models.GqlIndustryData,
)
def get_key_metrics_ttm(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlKeyMetricsTtm:
return self._transport.request(
"GET",
f"/v2/key-metrics-ttm/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlKeyMetricsTtm,
)
def get_market_calendar(
self,
*,
fields: str | None = None,
from_: str | None = None,
kind: models.MarketCalendarKindParam,
to: str | None = None,
) -> list[models.GqlMarketCalendarEntry]:
return self._transport.request(
"GET",
"/v2/market-calendar",
query=query_of({"fields": fields, "from": from_, "kind": kind, "to": to}),
into=list[models.GqlMarketCalendarEntry],
)
def get_market_summary(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
region: models.Region | None = None,
) -> list[models.GqlMarketSummaryQuote]:
return self._transport.request(
"GET",
"/v2/market-summary",
query=query_of({"fields": fields, "format": format, "lang": lang, "region": region}),
into=list[models.GqlMarketSummaryQuote],
)
def get_metrics(
self,
) -> models.MetricsText:
return self._transport.request(
"GET",
"/v2/metrics",
into=models.MetricsText,
)
def get_news(
self,
symbol: str,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
) -> list[models.GqlNews]:
return self._transport.request(
"GET",
f"/v2/news/{path_seg(symbol)}",
query=query_of({"count": count, "fields": fields, "lang": lang}),
into=list[models.GqlNews],
)
def iter_news(
self,
symbol: str,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
limit: int = 50,
) -> SyncPager[models.GqlNews]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/news/{path_seg(symbol)}",
query=query_of({"count": count, "fields": fields, "lang": lang, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, models.GqlNews)
def get_options(
self,
symbol: str,
*,
date: int | None = None,
fields: str | None = None,
) -> models.GqlOptions:
return self._transport.request(
"GET",
f"/v2/options/{path_seg(symbol)}",
query=query_of({"date": date, "fields": fields}),
into=models.GqlOptions,
)
def iter_options(
self,
symbol: str,
*,
date: int | None = None,
fields: str | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
f"/v2/options/{path_seg(symbol)}",
query=query_of({"date": date, "fields": fields, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_press_releases(
self,
symbol: str,
*,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlPressRelease]:
return self._transport.request(
"GET",
f"/v2/press-releases/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit}),
into=list[models.GqlPressRelease],
)
def get_price_target_consensus(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlPriceTargetConsensus:
return self._transport.request(
"GET",
f"/v2/price-target-consensus/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlPriceTargetConsensus,
)
def get_price_target_summary(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlPriceTargetSummary:
return self._transport.request(
"GET",
f"/v2/price-target-summary/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlPriceTargetSummary,
)
def get_protocol_tvl(
self,
id: str,
*,
fields: str | None = None,
) -> models.GqlProtocolTvl:
return self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}/tvl",
query=query_of({"fields": fields}),
into=models.GqlProtocolTvl,
)
def get_protocol_tvl_history(
self,
id: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlTvlPoint]:
return self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}/tvl-history",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlTvlPoint],
)
def get_quote(
self,
symbol: str,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
) -> models.GqlQuote:
return self._transport.request(
"GET",
f"/v2/quote/{path_seg(symbol)}",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo}),
into=models.GqlQuote,
)
def get_quote_type(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlQuoteTypeData:
return self._transport.request(
"GET",
f"/v2/quote-type/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlQuoteTypeData,
)
def get_quotes(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
symbols: str,
) -> models.GqlQuotesBatch:
return self._transport.request(
"GET",
"/v2/quotes",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo, "symbols": symbols}),
into=models.GqlQuotesBatch,
)
def iter_quotes(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
symbols: str,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/quotes",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
def get_rating_consensus(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlRatingConsensus:
return self._transport.request(
"GET",
f"/v2/rating-consensus/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlRatingConsensus,
)
def get_ratios_ttm(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlFinancialRatiosTtm:
return self._transport.request(
"GET",
f"/v2/ratios-ttm/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlFinancialRatiosTtm,
)
def get_recommendations(
self,
symbol: str,
*,
fields: str | None = None,
limit: int | None = None,
) -> models.GqlRecommendation:
return self._transport.request(
"GET",
f"/v2/recommendations/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit}),
into=models.GqlRecommendation,
)
def get_risk(
self,
symbol: str,
*,
benchmark: str | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
) -> models.GqlRiskSummary:
return self._transport.request(
"GET",
f"/v2/risk/{path_seg(symbol)}",
query=query_of({"benchmark": benchmark, "fields": fields, "interval": interval, "range": range}),
into=models.GqlRiskSummary,
)
def get_risk_factors(
self,
symbol: str,
*,
fields: str | None = None,
) -> list[models.GqlRiskFactor]:
return self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/risk-factors",
query=query_of({"fields": fields}),
into=list[models.GqlRiskFactor],
)
def get_screeners(
self,
screener: models.Screener,
*,
count: int | None = None,
fields: str | None = None,
format: models.ValueFormat | None = None,
) -> models.GqlScreenerResults:
return self._transport.request(
"GET",
f"/v2/screeners/{path_seg(screener)}",
query=query_of({"count": count, "fields": fields, "format": format}),
into=models.GqlScreenerResults,
)
def get_sector(
self,
sector: models.Sector,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
) -> models.GqlSectorData:
return self._transport.request(
"GET",
f"/v2/sectors/{path_seg(sector)}",
query=query_of({"fields": fields, "format": format, "lang": lang}),
into=models.GqlSectorData,
)
def get_sector_pe(
self,
*,
fields: str | None = None,
) -> list[models.GqlMarketSectorPe]:
return self._transport.request(
"GET",
"/v2/sector-pe",
query=query_of({"fields": fields}),
into=list[models.GqlMarketSectorPe],
)
def get_sector_performance(
self,
*,
fields: str | None = None,
) -> list[models.GqlMarketSectorPerformance]:
return self._transport.request(
"GET",
"/v2/sector-performance",
query=query_of({"fields": fields}),
into=list[models.GqlMarketSectorPerformance],
)
def get_sector_performance_history(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlSectorPerformanceHistory]:
return self._transport.request(
"GET",
"/v2/sector-performance/history",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlSectorPerformanceHistory],
)
def get_short_volume(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlShortVolume]:
return self._transport.request(
"GET",
f"/v2/short-volume/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlShortVolume],
)
def get_spark(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlSparkBatch:
return self._transport.request(
"GET",
"/v2/spark",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols}),
into=models.GqlSparkBatch,
)
def get_splits(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> list[models.GqlSplit]:
return self._transport.request(
"GET",
f"/v2/splits/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=list[models.GqlSplit],
)
def get_symbol_details(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlSymbolDetails:
return self._transport.request(
"GET",
f"/v2/symbol-details/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlSymbolDetails,
)
def get_transcript(
self,
symbol: str,
*,
lang: str | None = None,
quarter: models.Quarter | None = None,
year: int | None = None,
) -> models.GqlTranscript:
return self._transport.request(
"GET",
f"/v2/transcripts/{path_seg(symbol)}",
query=query_of({"lang": lang, "quarter": quarter, "year": year}),
into=models.GqlTranscript,
)
def get_transcripts(
self,
symbol: str,
*,
lang: str | None = None,
limit: int | None = None,
) -> list[models.GqlTranscriptWithMeta]:
return self._transport.request(
"GET",
f"/v2/transcripts/{path_seg(symbol)}/all",
query=query_of({"lang": lang, "limit": limit}),
into=list[models.GqlTranscriptWithMeta],
)
def get_trending(
self,
*,
fields: str | None = None,
region: models.Region | None = None,
) -> list[models.GqlTrendingQuote]:
return self._transport.request(
"GET",
"/v2/trending",
query=query_of({"fields": fields, "region": region}),
into=list[models.GqlTrendingQuote],
)
def health_check(
self,
) -> models.HealthResponse:
return self._transport.request(
"GET",
"/v2/health",
into=models.HealthResponse,
)
def lookup(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
logo: bool | None = None,
q: str,
region: models.Region | None = None,
type: models.LookupType | None = None,
) -> models.GqlLookupResults:
return self._transport.request(
"GET",
"/v2/lookup",
query=query_of({"count": count, "fields": fields, "lang": lang, "logo": logo, "q": q, "region": region, "type": type}),
into=models.GqlLookupResults,
)
def ping(
self,
) -> models.PingResponse:
return self._transport.request(
"GET",
"/v2/ping",
into=models.PingResponse,
)
def post_backtest(
self,
symbol: str,
body: models.BacktestRequest,
) -> models.GqlBacktestResult:
return self._transport.request(
"POST",
f"/v2/backtest/{path_seg(symbol)}",
json_body=to_wire(body),
into=models.GqlBacktestResult,
)
def post_custom_screener(
self,
body: models.CustomScreenerRequest,
) -> models.GqlScreenerResults:
return self._transport.request(
"POST",
"/v2/screeners/custom",
json_body=to_wire(body),
into=models.GqlScreenerResults,
)
def search(
self,
*,
cultural: bool | None = None,
fields: str | None = None,
fuzzy: bool | None = None,
lang: str | None = None,
logo: bool | None = None,
news: int | None = None,
q: str,
quotes: int | None = None,
region: models.Region | None = None,
research: bool | None = None,
) -> models.GqlSearchResults:
return self._transport.request(
"GET",
"/v2/search",
query=query_of({"cultural": cultural, "fields": fields, "fuzzy": fuzzy, "lang": lang, "logo": logo, "news": news, "q": q, "quotes": quotes, "region": region, "research": research}),
into=models.GqlSearchResults,
)
def iter_search(
self,
*,
cultural: bool | None = None,
fields: str | None = None,
fuzzy: bool | None = None,
lang: str | None = None,
logo: bool | None = None,
news: int | None = None,
q: str,
quotes: int | None = None,
region: models.Region | None = None,
research: bool | None = None,
limit: int = 50,
) -> SyncPager[typing.Any]:
def fetch(cursor: str | None) -> typing.Any:
return self._transport.request(
"GET",
"/v2/search",
query=query_of({"cultural": cultural, "fields": fields, "fuzzy": fuzzy, "lang": lang, "logo": logo, "news": news, "q": q, "quotes": quotes, "region": region, "research": research, "limit": limit, "cursor": cursor}),
)
return SyncPager(fetch, typing.Any)
class AsyncClient:
def __init__(
self,
base_url: str | None = None,
*,
server_env: ServerEnv | typing.Mapping[str, str] | None = None,
timeout: float = 30.0,
max_retries: int = 3,
on_rate_limit: typing.Callable[[typing.Any], None] | None = None,
transport: AsyncTransport | None = None,
) -> None:
base_url = base_url or embedded_base_url(EMBED, typing.cast("typing.Mapping[str, str] | None", server_env))
self._transport = transport or AsyncTransport(
base_url, timeout=timeout, max_retries=max_retries, on_rate_limit=on_rate_limit
)
async def aclose(self) -> None:
await self._transport.aclose()
async def __aenter__(self) -> AsyncClient:
return self
async def __aexit__(self, *exc: object) -> None:
await self.aclose()
async def get_analysis(
self,
symbol: str,
type: models.AnalysisType,
*,
fields: str | None = None,
) -> models.AnalysisResponse:
return await self._transport.request(
"GET",
f"/v2/analysis/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields}),
into=models.AnalysisResponse,
)
async def get_batch_capital_gains(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlCapitalGainsBatch:
return await self._transport.request(
"GET",
"/v2/capital-gains",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlCapitalGainsBatch,
)
async def get_batch_charts(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlChartsBatch:
return await self._transport.request(
"GET",
"/v2/charts",
query=query_of({"fields": fields, "interval": interval, "patterns": patterns, "range": range, "symbols": symbols}),
into=models.GqlChartsBatch,
)
def iter_batch_charts(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/charts",
query=query_of({"fields": fields, "interval": interval, "patterns": patterns, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_batch_dividends(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlDividendsBatch:
return await self._transport.request(
"GET",
"/v2/dividends",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlDividendsBatch,
)
def iter_batch_dividends(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/dividends",
query=query_of({"fields": fields, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_batch_financials(
self,
*,
fields: str | None = None,
frequency: models.Frequency | None = None,
metrics: str | None = None,
statement: models.StatementType,
symbols: str,
) -> models.GqlFinancialsBatch:
return await self._transport.request(
"GET",
"/v2/financials",
query=query_of({"fields": fields, "frequency": frequency, "metrics": metrics, "statement": statement, "symbols": symbols}),
into=models.GqlFinancialsBatch,
)
async def get_batch_indicators(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlIndicatorsBatch:
return await self._transport.request(
"GET",
"/v2/indicators",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols}),
into=models.GqlIndicatorsBatch,
)
def iter_batch_indicators(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/indicators",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_batch_options(
self,
*,
date: int | None = None,
fields: str | None = None,
symbols: str,
) -> models.GqlOptionsBatch:
return await self._transport.request(
"GET",
"/v2/options",
query=query_of({"date": date, "fields": fields, "symbols": symbols}),
into=models.GqlOptionsBatch,
)
def iter_batch_options(
self,
*,
date: int | None = None,
fields: str | None = None,
symbols: str,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/options",
query=query_of({"date": date, "fields": fields, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_batch_recommendations(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
symbols: str,
) -> models.GqlRecommendationsBatch:
return await self._transport.request(
"GET",
"/v2/recommendations",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit, "symbols": symbols}),
into=models.GqlRecommendationsBatch,
)
async def get_batch_splits(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlSplitsBatch:
return await self._transport.request(
"GET",
"/v2/splits",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=models.GqlSplitsBatch,
)
async def get_calendar(
self,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> list[models.GqlCalendarEvent]:
return await self._transport.request(
"GET",
"/v2/calendar",
query=query_of({"fields": fields, "range": range, "symbols": symbols}),
into=list[models.GqlCalendarEvent],
)
async def get_capital_gains(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> list[models.GqlCapitalGain]:
return await self._transport.request(
"GET",
f"/v2/capital-gains/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=list[models.GqlCapitalGain],
)
async def get_chart(
self,
symbol: str,
*,
end: int | None = None,
events: bool | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
start: int | None = None,
) -> models.GqlChart:
return await self._transport.request(
"GET",
f"/v2/chart/{path_seg(symbol)}",
query=query_of({"end": end, "events": events, "fields": fields, "interval": interval, "patterns": patterns, "range": range, "start": start}),
into=models.GqlChart,
)
def iter_chart(
self,
symbol: str,
*,
end: int | None = None,
events: bool | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
patterns: bool | None = None,
range: models.TimeRange | None = None,
start: int | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/chart/{path_seg(symbol)}",
query=query_of({"end": end, "events": events, "fields": fields, "interval": interval, "patterns": patterns, "range": range, "start": start, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_commitments_of_traders(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> models.GqlCommitmentsOfTraders:
return await self._transport.request(
"GET",
f"/v2/cftc/cot/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=models.GqlCommitmentsOfTraders,
)
async def get_commodity(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlCommodityQuote:
return await self._transport.request(
"GET",
f"/v2/commodities/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlCommodityQuote,
)
async def get_company_profile(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlCompanyProfile:
return await self._transport.request(
"GET",
f"/v2/company-profile/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlCompanyProfile,
)
async def get_congressional_trades(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlCongressionalTrade]:
return await self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/congressional-trades",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlCongressionalTrade],
)
async def get_crypto_coin(
self,
id: str,
*,
fields: str | None = None,
vs_currency: str | None = None,
) -> models.GqlCoinQuote:
return await self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}",
query=query_of({"fields": fields, "vsCurrency": vs_currency}),
into=models.GqlCoinQuote,
)
async def get_crypto_coins(
self,
*,
count: int | None = None,
fields: str | None = None,
vs_currency: str | None = None,
) -> list[models.GqlCoinQuote]:
return await self._transport.request(
"GET",
"/v2/crypto/coins",
query=query_of({"count": count, "fields": fields, "vsCurrency": vs_currency}),
into=list[models.GqlCoinQuote],
)
def iter_crypto_coins(
self,
*,
count: int | None = None,
fields: str | None = None,
vs_currency: str | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlCoinQuote]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/crypto/coins",
query=query_of({"count": count, "fields": fields, "vsCurrency": vs_currency, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlCoinQuote)
async def get_crypto_global(
self,
*,
fields: str | None = None,
) -> models.GqlGlobalCryptoStats:
return await self._transport.request(
"GET",
"/v2/crypto/global",
query=query_of({"fields": fields}),
into=models.GqlGlobalCryptoStats,
)
async def get_crypto_news(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
) -> list[models.GqlNews]:
return await self._transport.request(
"GET",
"/v2/crypto/news",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit}),
into=list[models.GqlNews],
)
async def get_crypto_search(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
limit_results: int | None = None,
query: str,
) -> list[models.GqlSymbolMatch]:
return await self._transport.request(
"GET",
"/v2/crypto/search",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit, "limitResults": limit_results, "query": query}),
into=list[models.GqlSymbolMatch],
)
async def get_crypto_trending(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlTrendingCoin]:
return await self._transport.request(
"GET",
"/v2/crypto/trending",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlTrendingCoin],
)
async def get_currencies(
self,
*,
fields: str | None = None,
) -> list[models.GqlCurrency]:
return await self._transport.request(
"GET",
"/v2/currencies",
query=query_of({"fields": fields}),
into=list[models.GqlCurrency],
)
async def get_dividends(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> models.GqlDividends:
return await self._transport.request(
"GET",
f"/v2/dividends/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=models.GqlDividends,
)
def iter_dividends(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/dividends/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_earnings_surprises(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEarningsSurpriseHistory:
return await self._transport.request(
"GET",
f"/v2/earnings-surprises/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEarningsSurpriseHistory,
)
async def get_earnings_transcript(
self,
symbol: str,
*,
fields: str | None = None,
quarter: models.Quarter | None = None,
year: int | None = None,
) -> models.GqlEarningsTranscript:
return await self._transport.request(
"GET",
f"/v2/earnings-transcript/{path_seg(symbol)}",
query=query_of({"fields": fields, "quarter": quarter, "year": year}),
into=models.GqlEarningsTranscript,
)
async def get_edgar_cik(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEdgarCik:
return await self._transport.request(
"GET",
f"/v2/edgar/cik/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEdgarCik,
)
async def get_edgar_facts(
self,
symbol: str,
*,
concepts: str | None = None,
fields: str | None = None,
taxonomy: str | None = None,
) -> list[models.GqlFactConcept]:
return await self._transport.request(
"GET",
f"/v2/edgar/facts/{path_seg(symbol)}",
query=query_of({"concepts": concepts, "fields": fields, "taxonomy": taxonomy}),
into=list[models.GqlFactConcept],
)
def iter_edgar_facts(
self,
symbol: str,
*,
concepts: str | None = None,
fields: str | None = None,
taxonomy: str | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlFactConcept]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/edgar/facts/{path_seg(symbol)}",
query=query_of({"concepts": concepts, "fields": fields, "taxonomy": taxonomy, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlFactConcept)
async def get_edgar_search(
self,
*,
end_date: str | None = None,
fields: str | None = None,
forms: str | None = None,
from_: int | None = None,
q: str,
size: int | None = None,
start_date: str | None = None,
) -> models.GqlEdgarSearchResults:
return await self._transport.request(
"GET",
"/v2/edgar/search",
query=query_of({"endDate": end_date, "fields": fields, "forms": forms, "from": from_, "q": q, "size": size, "startDate": start_date}),
into=models.GqlEdgarSearchResults,
)
async def get_edgar_submissions(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEdgarSubmissions:
return await self._transport.request(
"GET",
f"/v2/edgar/submissions/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEdgarSubmissions,
)
def iter_edgar_submissions(
self,
symbol: str,
*,
fields: str | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/edgar/submissions/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_etf_profile(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlEtfProfile:
return await self._transport.request(
"GET",
f"/v2/etf-profile/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlEtfProfile,
)
async def get_exchanges(
self,
*,
fields: str | None = None,
) -> list[models.GqlExchange]:
return await self._transport.request(
"GET",
"/v2/exchanges",
query=query_of({"fields": fields}),
into=list[models.GqlExchange],
)
async def get_executive_compensation(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlExecutiveCompensation]:
return await self._transport.request(
"GET",
f"/v2/executive-compensation/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlExecutiveCompensation],
)
async def get_fails_to_deliver(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlFailToDeliver]:
return await self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/fails-to-deliver",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlFailToDeliver],
)
async def get_fear_and_greed(
self,
*,
fields: str | None = None,
) -> models.GqlFearAndGreed:
return await self._transport.request(
"GET",
"/v2/fear-and-greed",
query=query_of({"fields": fields}),
into=models.GqlFearAndGreed,
)
async def get_feeds(
self,
*,
fields: str | None = None,
form_type: str | None = None,
sources: str | None = None,
) -> list[models.GqlFeedEntry]:
return await self._transport.request(
"GET",
"/v2/feeds",
query=query_of({"fields": fields, "formType": form_type, "sources": sources}),
into=list[models.GqlFeedEntry],
)
def iter_feeds(
self,
*,
fields: str | None = None,
form_type: str | None = None,
sources: str | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlFeedEntry]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/feeds",
query=query_of({"fields": fields, "formType": form_type, "sources": sources, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlFeedEntry)
async def get_filing_sections(
self,
symbol: str,
*,
accession_number: str,
fields: str | None = None,
form: models.FilingSectionFormParam,
) -> list[models.GqlFilingSection]:
return await self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/sections",
query=query_of({"accessionNumber": accession_number, "fields": fields, "form": form}),
into=list[models.GqlFilingSection],
)
async def get_financials(
self,
symbol: str,
statement: models.StatementType,
*,
fields: str | None = None,
frequency: models.Frequency | None = None,
metrics: str | None = None,
) -> list[models.GqlFinancialLineItem]:
return await self._transport.request(
"GET",
f"/v2/financials/{path_seg(symbol)}/{path_seg(statement)}",
query=query_of({"fields": fields, "frequency": frequency, "metrics": metrics}),
into=list[models.GqlFinancialLineItem],
)
async def get_forex(
self,
from_: str,
to: str,
*,
fields: str | None = None,
) -> models.GqlForexQuote:
return await self._transport.request(
"GET",
f"/v2/forex/{path_seg(from_)}/{path_seg(to)}",
query=query_of({"fields": fields}),
into=models.GqlForexQuote,
)
async def get_forex_news(
self,
*,
fields: str | None = None,
limit: int | None = None,
page_cursor: str | None = None,
page_limit: int | None = None,
) -> list[models.GqlNews]:
return await self._transport.request(
"GET",
"/v2/forex/news",
query=query_of({"fields": fields, "limit": limit, "pageCursor": page_cursor, "pageLimit": page_limit}),
into=list[models.GqlNews],
)
async def get_fred_series(
self,
id: str,
*,
fields: str | None = None,
) -> models.GqlMacroSeries:
return await self._transport.request(
"GET",
f"/v2/fred/series/{path_seg(id)}",
query=query_of({"fields": fields}),
into=models.GqlMacroSeries,
)
def iter_fred_series(
self,
id: str,
*,
fields: str | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/fred/series/{path_seg(id)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_fred_treasury_yields(
self,
*,
fields: str | None = None,
year: int | None = None,
) -> list[models.GqlTreasuryYield]:
return await self._transport.request(
"GET",
"/v2/fred/treasury-yields",
query=query_of({"fields": fields, "year": year}),
into=list[models.GqlTreasuryYield],
)
def iter_fred_treasury_yields(
self,
*,
fields: str | None = None,
year: int | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlTreasuryYield]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/fred/treasury-yields",
query=query_of({"fields": fields, "year": year, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlTreasuryYield)
async def get_futures(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlFuturesQuote:
return await self._transport.request(
"GET",
f"/v2/futures/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlFuturesQuote,
)
async def get_gdelt_news(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlNews]:
return await self._transport.request(
"GET",
f"/v2/gdelt/news/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlNews],
)
async def get_general_news(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
) -> list[models.GqlNews]:
return await self._transport.request(
"GET",
"/v2/news",
query=query_of({"count": count, "fields": fields, "lang": lang}),
into=list[models.GqlNews],
)
def iter_general_news(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlNews]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/news",
query=query_of({"count": count, "fields": fields, "lang": lang, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlNews)
async def get_grading_actions(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlGradingAction]:
return await self._transport.request(
"GET",
f"/v2/grading-actions/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlGradingAction],
)
async def get_holders(
self,
symbol: str,
type: models.HolderType,
*,
fields: str | None = None,
) -> models.HoldersResponse:
return await self._transport.request(
"GET",
f"/v2/holders/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields}),
into=models.HoldersResponse,
)
def iter_holders(
self,
symbol: str,
type: models.HolderType,
*,
fields: str | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/holders/{path_seg(symbol)}/{path_seg(type)}",
query=query_of({"fields": fields, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_hours(
self,
*,
fields: str | None = None,
region: models.Region | None = None,
) -> models.GqlMarketHours:
return await self._transport.request(
"GET",
"/v2/hours",
query=query_of({"fields": fields, "region": region}),
into=models.GqlMarketHours,
)
async def get_index_constituent_changes(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlIndexConstituentChange]:
return await self._transport.request(
"GET",
f"/v2/indices/{path_seg(symbol)}/constituent-changes",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlIndexConstituentChange],
)
async def get_index_constituents(
self,
symbol: str,
*,
fields: str | None = None,
) -> list[models.GqlIndexConstituent]:
return await self._transport.request(
"GET",
f"/v2/index-constituents/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=list[models.GqlIndexConstituent],
)
async def get_indicators(
self,
symbol: str,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
) -> models.GqlIndicatorsSummary:
return await self._transport.request(
"GET",
f"/v2/indicators/{path_seg(symbol)}",
query=query_of({"fields": fields, "interval": interval, "range": range}),
into=models.GqlIndicatorsSummary,
)
async def get_indices(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
region: models.indices_Region | None = None,
) -> list[models.GqlQuote]:
return await self._transport.request(
"GET",
"/v2/indices",
query=query_of({"fields": fields, "format": format, "region": region}),
into=list[models.GqlQuote],
)
async def get_industry(
self,
industry: models.Industry,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
) -> models.GqlIndustryData:
return await self._transport.request(
"GET",
f"/v2/industries/{path_seg(industry)}",
query=query_of({"fields": fields, "format": format, "lang": lang}),
into=models.GqlIndustryData,
)
async def get_key_metrics_ttm(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlKeyMetricsTtm:
return await self._transport.request(
"GET",
f"/v2/key-metrics-ttm/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlKeyMetricsTtm,
)
async def get_market_calendar(
self,
*,
fields: str | None = None,
from_: str | None = None,
kind: models.MarketCalendarKindParam,
to: str | None = None,
) -> list[models.GqlMarketCalendarEntry]:
return await self._transport.request(
"GET",
"/v2/market-calendar",
query=query_of({"fields": fields, "from": from_, "kind": kind, "to": to}),
into=list[models.GqlMarketCalendarEntry],
)
async def get_market_summary(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
region: models.Region | None = None,
) -> list[models.GqlMarketSummaryQuote]:
return await self._transport.request(
"GET",
"/v2/market-summary",
query=query_of({"fields": fields, "format": format, "lang": lang, "region": region}),
into=list[models.GqlMarketSummaryQuote],
)
async def get_metrics(
self,
) -> models.MetricsText:
return await self._transport.request(
"GET",
"/v2/metrics",
into=models.MetricsText,
)
async def get_news(
self,
symbol: str,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
) -> list[models.GqlNews]:
return await self._transport.request(
"GET",
f"/v2/news/{path_seg(symbol)}",
query=query_of({"count": count, "fields": fields, "lang": lang}),
into=list[models.GqlNews],
)
def iter_news(
self,
symbol: str,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
limit: int = 50,
) -> AsyncPager[models.GqlNews]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/news/{path_seg(symbol)}",
query=query_of({"count": count, "fields": fields, "lang": lang, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, models.GqlNews)
async def get_options(
self,
symbol: str,
*,
date: int | None = None,
fields: str | None = None,
) -> models.GqlOptions:
return await self._transport.request(
"GET",
f"/v2/options/{path_seg(symbol)}",
query=query_of({"date": date, "fields": fields}),
into=models.GqlOptions,
)
def iter_options(
self,
symbol: str,
*,
date: int | None = None,
fields: str | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
f"/v2/options/{path_seg(symbol)}",
query=query_of({"date": date, "fields": fields, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_press_releases(
self,
symbol: str,
*,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlPressRelease]:
return await self._transport.request(
"GET",
f"/v2/press-releases/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit}),
into=list[models.GqlPressRelease],
)
async def get_price_target_consensus(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlPriceTargetConsensus:
return await self._transport.request(
"GET",
f"/v2/price-target-consensus/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlPriceTargetConsensus,
)
async def get_price_target_summary(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlPriceTargetSummary:
return await self._transport.request(
"GET",
f"/v2/price-target-summary/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlPriceTargetSummary,
)
async def get_protocol_tvl(
self,
id: str,
*,
fields: str | None = None,
) -> models.GqlProtocolTvl:
return await self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}/tvl",
query=query_of({"fields": fields}),
into=models.GqlProtocolTvl,
)
async def get_protocol_tvl_history(
self,
id: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlTvlPoint]:
return await self._transport.request(
"GET",
f"/v2/crypto/coins/{path_seg(id)}/tvl-history",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlTvlPoint],
)
async def get_quote(
self,
symbol: str,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
) -> models.GqlQuote:
return await self._transport.request(
"GET",
f"/v2/quote/{path_seg(symbol)}",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo}),
into=models.GqlQuote,
)
async def get_quote_type(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlQuoteTypeData:
return await self._transport.request(
"GET",
f"/v2/quote-type/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlQuoteTypeData,
)
async def get_quotes(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
symbols: str,
) -> models.GqlQuotesBatch:
return await self._transport.request(
"GET",
"/v2/quotes",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo, "symbols": symbols}),
into=models.GqlQuotesBatch,
)
def iter_quotes(
self,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
logo: bool | None = None,
symbols: str,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/quotes",
query=query_of({"fields": fields, "format": format, "lang": lang, "logo": logo, "symbols": symbols, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)
async def get_rating_consensus(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlRatingConsensus:
return await self._transport.request(
"GET",
f"/v2/rating-consensus/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlRatingConsensus,
)
async def get_ratios_ttm(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlFinancialRatiosTtm:
return await self._transport.request(
"GET",
f"/v2/ratios-ttm/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlFinancialRatiosTtm,
)
async def get_recommendations(
self,
symbol: str,
*,
fields: str | None = None,
limit: int | None = None,
) -> models.GqlRecommendation:
return await self._transport.request(
"GET",
f"/v2/recommendations/{path_seg(symbol)}",
query=query_of({"fields": fields, "limit": limit}),
into=models.GqlRecommendation,
)
async def get_risk(
self,
symbol: str,
*,
benchmark: str | None = None,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
) -> models.GqlRiskSummary:
return await self._transport.request(
"GET",
f"/v2/risk/{path_seg(symbol)}",
query=query_of({"benchmark": benchmark, "fields": fields, "interval": interval, "range": range}),
into=models.GqlRiskSummary,
)
async def get_risk_factors(
self,
symbol: str,
*,
fields: str | None = None,
) -> list[models.GqlRiskFactor]:
return await self._transport.request(
"GET",
f"/v2/filings/{path_seg(symbol)}/risk-factors",
query=query_of({"fields": fields}),
into=list[models.GqlRiskFactor],
)
async def get_screeners(
self,
screener: models.Screener,
*,
count: int | None = None,
fields: str | None = None,
format: models.ValueFormat | None = None,
) -> models.GqlScreenerResults:
return await self._transport.request(
"GET",
f"/v2/screeners/{path_seg(screener)}",
query=query_of({"count": count, "fields": fields, "format": format}),
into=models.GqlScreenerResults,
)
async def get_sector(
self,
sector: models.Sector,
*,
fields: str | None = None,
format: models.ValueFormat | None = None,
lang: str | None = None,
) -> models.GqlSectorData:
return await self._transport.request(
"GET",
f"/v2/sectors/{path_seg(sector)}",
query=query_of({"fields": fields, "format": format, "lang": lang}),
into=models.GqlSectorData,
)
async def get_sector_pe(
self,
*,
fields: str | None = None,
) -> list[models.GqlMarketSectorPe]:
return await self._transport.request(
"GET",
"/v2/sector-pe",
query=query_of({"fields": fields}),
into=list[models.GqlMarketSectorPe],
)
async def get_sector_performance(
self,
*,
fields: str | None = None,
) -> list[models.GqlMarketSectorPerformance]:
return await self._transport.request(
"GET",
"/v2/sector-performance",
query=query_of({"fields": fields}),
into=list[models.GqlMarketSectorPerformance],
)
async def get_sector_performance_history(
self,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlSectorPerformanceHistory]:
return await self._transport.request(
"GET",
"/v2/sector-performance/history",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlSectorPerformanceHistory],
)
async def get_short_volume(
self,
symbol: str,
*,
cursor: str | None = None,
fields: str | None = None,
limit: int | None = None,
) -> list[models.GqlShortVolume]:
return await self._transport.request(
"GET",
f"/v2/short-volume/{path_seg(symbol)}",
query=query_of({"cursor": cursor, "fields": fields, "limit": limit}),
into=list[models.GqlShortVolume],
)
async def get_spark(
self,
*,
fields: str | None = None,
interval: models.Interval | None = None,
range: models.TimeRange | None = None,
symbols: str,
) -> models.GqlSparkBatch:
return await self._transport.request(
"GET",
"/v2/spark",
query=query_of({"fields": fields, "interval": interval, "range": range, "symbols": symbols}),
into=models.GqlSparkBatch,
)
async def get_splits(
self,
symbol: str,
*,
fields: str | None = None,
range: models.TimeRange | None = None,
) -> list[models.GqlSplit]:
return await self._transport.request(
"GET",
f"/v2/splits/{path_seg(symbol)}",
query=query_of({"fields": fields, "range": range}),
into=list[models.GqlSplit],
)
async def get_symbol_details(
self,
symbol: str,
*,
fields: str | None = None,
) -> models.GqlSymbolDetails:
return await self._transport.request(
"GET",
f"/v2/symbol-details/{path_seg(symbol)}",
query=query_of({"fields": fields}),
into=models.GqlSymbolDetails,
)
async def get_transcript(
self,
symbol: str,
*,
lang: str | None = None,
quarter: models.Quarter | None = None,
year: int | None = None,
) -> models.GqlTranscript:
return await self._transport.request(
"GET",
f"/v2/transcripts/{path_seg(symbol)}",
query=query_of({"lang": lang, "quarter": quarter, "year": year}),
into=models.GqlTranscript,
)
async def get_transcripts(
self,
symbol: str,
*,
lang: str | None = None,
limit: int | None = None,
) -> list[models.GqlTranscriptWithMeta]:
return await self._transport.request(
"GET",
f"/v2/transcripts/{path_seg(symbol)}/all",
query=query_of({"lang": lang, "limit": limit}),
into=list[models.GqlTranscriptWithMeta],
)
async def get_trending(
self,
*,
fields: str | None = None,
region: models.Region | None = None,
) -> list[models.GqlTrendingQuote]:
return await self._transport.request(
"GET",
"/v2/trending",
query=query_of({"fields": fields, "region": region}),
into=list[models.GqlTrendingQuote],
)
async def health_check(
self,
) -> models.HealthResponse:
return await self._transport.request(
"GET",
"/v2/health",
into=models.HealthResponse,
)
async def lookup(
self,
*,
count: int | None = None,
fields: str | None = None,
lang: str | None = None,
logo: bool | None = None,
q: str,
region: models.Region | None = None,
type: models.LookupType | None = None,
) -> models.GqlLookupResults:
return await self._transport.request(
"GET",
"/v2/lookup",
query=query_of({"count": count, "fields": fields, "lang": lang, "logo": logo, "q": q, "region": region, "type": type}),
into=models.GqlLookupResults,
)
async def ping(
self,
) -> models.PingResponse:
return await self._transport.request(
"GET",
"/v2/ping",
into=models.PingResponse,
)
async def post_backtest(
self,
symbol: str,
body: models.BacktestRequest,
) -> models.GqlBacktestResult:
return await self._transport.request(
"POST",
f"/v2/backtest/{path_seg(symbol)}",
json_body=to_wire(body),
into=models.GqlBacktestResult,
)
async def post_custom_screener(
self,
body: models.CustomScreenerRequest,
) -> models.GqlScreenerResults:
return await self._transport.request(
"POST",
"/v2/screeners/custom",
json_body=to_wire(body),
into=models.GqlScreenerResults,
)
async def search(
self,
*,
cultural: bool | None = None,
fields: str | None = None,
fuzzy: bool | None = None,
lang: str | None = None,
logo: bool | None = None,
news: int | None = None,
q: str,
quotes: int | None = None,
region: models.Region | None = None,
research: bool | None = None,
) -> models.GqlSearchResults:
return await self._transport.request(
"GET",
"/v2/search",
query=query_of({"cultural": cultural, "fields": fields, "fuzzy": fuzzy, "lang": lang, "logo": logo, "news": news, "q": q, "quotes": quotes, "region": region, "research": research}),
into=models.GqlSearchResults,
)
def iter_search(
self,
*,
cultural: bool | None = None,
fields: str | None = None,
fuzzy: bool | None = None,
lang: str | None = None,
logo: bool | None = None,
news: int | None = None,
q: str,
quotes: int | None = None,
region: models.Region | None = None,
research: bool | None = None,
limit: int = 50,
) -> AsyncPager[typing.Any]:
async def fetch(cursor: str | None) -> typing.Any:
return await self._transport.request(
"GET",
"/v2/search",
query=query_of({"cultural": cultural, "fields": fields, "fuzzy": fuzzy, "lang": lang, "logo": logo, "news": news, "q": q, "quotes": quotes, "region": region, "research": research, "limit": limit, "cursor": cursor}),
)
return AsyncPager(fetch, typing.Any)