Skip to content

API.md

Latest commit

 

History

History
2062 lines (1243 loc) · 54.2 KB

API.md

File metadata and controls

2062 lines (1243 loc) · 54.2 KB

velo_sdk.api

velo_sdk.api.errors

APIError Objects

class APIError(Exception)

General exception for API errors.

Attributes:

  • message - The error message.
  • code - HTTP status code.
  • status - HTTP status text.
  • timestamp - When the error occurred.

RateLimitError Objects

class RateLimitError(APIError)

Exception raised when rate limiting is exceeded.

InsufficientCreditsError Objects

class InsufficientCreditsError(APIError)

Exception raised when the user has insufficient API credits. This is a non-retryable error that requires the user to add credits.

velo_sdk.api.climate

Climate Objects

class Climate()

list_horizons

def list_horizons() -> list[int]

List the available horizons for climate analysis.

list_pathways

def list_pathways() -> list[Pathway]

List the available pathways for climate analysis.

velo_sdk.api.base

BaseClient Objects

class BaseClient()

get

def get(path: str, params: Dict[str, Any] = dict(),
        **kwargs) -> Dict[str, Any]

Make a GET request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • params - Query parameters
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

get_async

async def get_async(path: str, params: Dict[str, Any] = dict(),
                    **kwargs) -> Dict[str, Any]

Make an asynchronous GET request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • params - Query parameters
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

post

def post(path: str, json: Dict[str, Any] = dict(), **kwargs) -> Dict[str, Any]

Make a POST request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

post_async

async def post_async(path: str, json: Dict[str, Any] = dict(),
                     **kwargs) -> Dict[str, Any]

Make an asynchronous POST request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

put

def put(path: str, json: Dict[str, Any] = dict(), **kwargs) -> Dict[str, Any]

Make a PUT request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

put_async

async def put_async(path: str, json: Dict[str, Any] = dict(),
                    **kwargs) -> Dict[str, Any]

Make an asynchronous PUT request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

delete

def delete(path: str, **kwargs) -> Dict[str, Any]

Make a DELETE request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

delete_async

async def delete_async(path: str, **kwargs) -> Dict[str, Any]

Make an asynchronous DELETE request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

patch

def patch(path: str, json: Dict[str, Any] = dict(),
          **kwargs) -> Dict[str, Any]

Make a PATCH request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

patch_async

async def patch_async(path: str, json: Dict[str, Any] = dict(),
                      **kwargs) -> Dict[str, Any]

Make an asynchronous PATCH request to the API. Version prefix /v3 is added only when the endpoint has no version prefix. Paths like /v3/assets or /v4/climate/metrics are used as-is. Absolute URLs are also supported and used as-is.

Arguments:

  • path - API endpoint path
  • json - JSON body data
  • **kwargs - Additional arguments to pass to the request

Returns:

API response data as JSON deserialized into a dictionary

velo_sdk.api.pagination

PaginatedIterator Objects

class PaginatedIterator(Generic[T], Iterator[T])

fetch_page

def fetch_page() -> list[T]

Fetches the next page of results and returns them as a list.

to_polars

def to_polars()

Fetches all items from all pages, applies an optional transformation, and returns them as a Polars DataFrame. This method will consume the iterator.

Raises:

  • ImportError - If Polars is not installed. Install it with: pip install velo-sdk[polars]

AsyncPaginatedIterator Objects

class AsyncPaginatedIterator(Generic[T], AsyncIterator[T])

afetch_page

async def afetch_page() -> list[T]

Asynchronously fetches the next page of results and returns them as a list.

to_polars

async def to_polars()

Asynchronously fetches all items from all pages, applies an optional transformation, and returns them as a Polars DataFrame. This method will consume the iterator.

Raises:

  • ImportError - If Polars is not installed. Install it with: pip install velo-sdk[polars]

velo_sdk.api.static_list

StaticListIterator Objects

class StaticListIterator(Generic[T], Iterator[T], AsyncIterator[T])

Helper to fetch, parse, and iterate over a list of items from an API endpoint that returns the full list under a 'results' key without pagination.

fetch_all

def fetch_all() -> List[T]

Synchronously fetches all results from the endpoint and parses them. If data has already been fetched for iteration, returns the cached data.

Returns:

A list of parsed items of type T.

__iter__

def __iter__() -> Self

Returns the iterator object itself, fetching data if needed.

__next__

def __next__() -> T

Returns the next item in the fetched list.

afetch_all

async def afetch_all() -> List[T]

Asynchronously fetches all results from the endpoint and parses them. If data has already been fetched for iteration, returns the cached data.

Returns:

A list of parsed items of type T.

__aiter__

def __aiter__() -> Self

Returns the async iterator object itself.

__anext__

async def __anext__() -> T

Returns the next item, fetching data asynchronously if needed.

to_polars

def to_polars()

Fetches all results (if not already fetched) and converts them into a Polars DataFrame.

Assumes that the generic type T is a Pydantic model.

Returns:

A Polars DataFrame containing the fetched data.

Raises:

  • ImportError - If Polars is not installed. Install it with: pip install velo-sdk[polars]

velo_sdk.api.companies

Companies Objects

class Companies()

get_company

def get_company(company_id: str) -> Company

Get a company by its unique ID.

Arguments:

  • company_id str - The unique identifier of the company.

Returns:

  • Company - The Company object.

get_company_async

async def get_company_async(company_id: str) -> Company

Get a company by its unique ID asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.

Returns:

  • Company - The Company object.

list_companies

def list_companies(*,
                   scope: Literal["public", "organization"] = "public",
                   **extra_params: Any) -> PaginatedIterator[Company]

List all companies.

Arguments:

  • scope Literal["public", "organization"] - The scope to filter companies by "public" is the default scope and searches all available companies in VELO. "organization" searches all private companies uploaded to the organization.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[Company] - An iterator over Company objects.

list_companies_async

async def list_companies_async(
        *,
        scope: Literal["public", "organization"] = "public",
        **extra_params: Any) -> AsyncPaginatedIterator[Company]

List all companies asynchronously.

Arguments:

  • scope Literal["public", "organization"] - The scope to filter companies by "public" is the default scope and searches all available companies in VELO. "organization" searches all private companies uploaded to the organization.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[Company] - An asynchronous iterator over Company objects.

search_companies

def search_companies(*,
                     name: Optional[str] = None,
                     **extra_params: Any) -> list[Company]

Search for companies by name.

Arguments:

  • name Optional[str] - The name of the company to search for.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • list[Company] - A list of Company objects matching the search criteria.

search_companies_async

async def search_companies_async(*,
                                 name: Optional[str] = None,
                                 **extra_params: Any) -> list[Company]

Search for companies by name asynchronously.

Arguments:

  • name Optional[str] - The name of the company to search for.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • list[Company] - A list of Company objects matching the search criteria.

list_company_assets

def list_company_assets(company_id: str,
                        **extra_params: Any) -> PaginatedIterator[Asset]

List all assets for a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[Asset] - An iterator over Asset objects belonging to the company.

list_company_assets_async

async def list_company_assets_async(
        company_id: str, **extra_params: Any) -> AsyncPaginatedIterator[Asset]

List all assets for a company asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[Asset] - An asynchronous iterator over Asset objects belonging to the company.

list_uninsurable_company_assets

def list_uninsurable_company_assets(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetClimateScore]

List all uninsurable assets for a company. Uninsurable assets are defined as those with cvar_95 >= 0.35.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetClimateScore] - An iterator over AssetClimateScore objects for uninsurable assets.

list_uninsurable_company_assets_async

async def list_uninsurable_company_assets_async(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetClimateScore]

List all uninsurable assets for a company asynchronously. Uninsurable assets are defined as those with cvar_95 >= 0.35.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetClimateScore] - An asynchronous iterator over AssetClimateScore objects for uninsurable assets.

list_stranded_company_assets

def list_stranded_company_assets(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetClimateScore]

List all stranded assets for a company. Stranded assets are defined as those with cvar_95 >= 0.75.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetClimateScore] - An iterator over AssetClimateScore objects for stranded assets.

list_stranded_company_assets_async

async def list_stranded_company_assets_async(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetClimateScore]

List all stranded assets for a company asynchronously. Stranded assets are defined as those with cvar_95 >= 0.75.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetClimateScore] - An asynchronous iterator over AssetClimateScore objects for stranded assets.

get_company_climate_scores

def get_company_climate_scores(company_id: str, pathway: Pathway,
                               horizon: HorizonYear) -> ClimateScore

Get the climate scores for a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • ClimateScore - The ClimateScore object for the company.

get_company_climate_scores_async

async def get_company_climate_scores_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> ClimateScore

Get the climate scores for a company asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • ClimateScore - The ClimateScore object for the company.

get_company_impact_scores

def get_company_impact_scores(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[ImpactScore]

Get the impact scores for a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[ImpactScore] - An iterator over ImpactScore objects for the company.

get_company_impact_scores_async

async def get_company_impact_scores_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[ImpactScore]

Get the impact scores for a company asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[ImpactScore] - An asynchronous iterator over ImpactScore objects for the company.

list_company_asset_climate_scores

def list_company_asset_climate_scores(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetClimateScore]

Get the climate scores for all assets of a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetClimateScore] - An iterator over AssetClimateScore objects for the company's assets.

list_company_asset_climate_scores_async

async def list_company_asset_climate_scores_async(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetClimateScore]

Get the climate scores for all assets of a company asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetClimateScore] - An asynchronous iterator over AssetClimateScore objects for the company's assets.

list_company_asset_impact_scores

def list_company_asset_impact_scores(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetImpactScore]

Get the impact scores for all assets of a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetImpactScore] - An iterator over AssetImpactScore objects for the company's assets.

list_company_asset_impact_scores_async

async def list_company_asset_impact_scores_async(
        company_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetImpactScore]

Get the impact scores for all assets of a company asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetImpactScore] - An asynchronous iterator over AssetImpactScore objects for the company's assets.

aggregate_company_asset_climate_scores_by_country

def aggregate_company_asset_climate_scores_by_country(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryClimateScore]

Get the climate scores for all assets of a company aggregated by country.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryClimateScore] - An iterator over CountryClimateScore objects, aggregated by country.

aggregate_company_asset_climate_scores_by_country_async

async def aggregate_company_asset_climate_scores_by_country_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryClimateScore]

Get the climate scores for all assets of a company aggregated by country asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryClimateScore] - An asynchronous iterator over CountryClimateScore objects, aggregated by country.

aggregate_company_asset_impact_scores_by_country

def aggregate_company_asset_impact_scores_by_country(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryImpactScore]

Get the impact scores for all assets of a company aggregated by country.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryImpactScore] - An iterator over CountryImpactScore objects, aggregated by country.

aggregate_company_asset_impact_scores_by_country_async

async def aggregate_company_asset_impact_scores_by_country_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryImpactScore]

Get the impact scores for all assets of a company aggregated by country asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryImpactScore] - An asynchronous iterator over CountryImpactScore objects, aggregated by country.

aggregate_company_asset_climate_scores_by_asset_type

def aggregate_company_asset_climate_scores_by_asset_type(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeClimateScore]

Get the climate scores for all assets of a company aggregated by asset type.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeClimateScore] - An iterator over AssetTypeClimateScore objects, aggregated by asset type.

aggregate_company_asset_climate_scores_by_asset_type_async

async def aggregate_company_asset_climate_scores_by_asset_type_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeClimateScore]

Get the climate scores for all assets of a company aggregated by asset type asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeClimateScore] - An asynchronous iterator over AssetTypeClimateScore objects, aggregated by asset type.

aggregate_company_asset_impact_scores_by_asset_type

def aggregate_company_asset_impact_scores_by_asset_type(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeImpactScore]

Get the impact scores for all assets of a company aggregated by asset type.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeImpactScore] - An iterator over AssetTypeImpactScore objects, aggregated by asset type.

aggregate_company_asset_impact_scores_by_asset_type_async

async def aggregate_company_asset_impact_scores_by_asset_type_async(
        company_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeImpactScore]

Get the impact scores for all assets of a company aggregated by asset type asynchronously.

Arguments:

  • company_id str - The unique identifier of the company.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeImpactScore] - An asynchronous iterator over AssetTypeImpactScore objects, aggregated by asset type.

upload_company_assets

def upload_company_assets(company_id: str, assets: list[Dict] | Any)

Upload new assets to a company.

Arguments:

  • company_id str - The unique identifier of the company.
  • assets list[Dict] | pl.DataFrame | pandas.DataFrame - A list of Asset objects to upload, or a DataFrame (pandas or polars) that will be converted to a list of dictionaries.

create_company

def create_company(name: str, sector: Sector) -> Company

Create a new company.

Arguments:

  • name str - The name of the company.
  • sector Sector - The sector of the company.

Returns:

  • Company - The newly created Company object.

create_company_async

async def create_company_async(name: str, sector: Sector) -> Company

Create a new company asynchronously.

Arguments:

  • name str - The name of the company.
  • sector Sector - The GICS sector of the company.

Returns:

  • Company - The newly created Company object.

velo_sdk.api.assets

Assets Objects

class Assets()

get_asset

def get_asset(asset_id: str) -> Asset

Get an asset by its unique ID.

Arguments:

  • asset_id str - The unique identifier of the asset.

Returns:

  • Asset - The Asset object.

get_asset_async

async def get_asset_async(asset_id: str) -> Asset

Get an asset by its unique ID asynchronously.

Arguments:

  • asset_id str - The unique identifier of the asset.

Returns:

  • Asset - The Asset object.

list_assets

def list_assets(**extra_params: Any) -> PaginatedIterator[Asset]

List all assets.

Arguments:

  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[Asset] - An iterator over Asset objects.

list_assets_async

async def list_assets_async(**extra_params: Any
                            ) -> AsyncPaginatedIterator[Asset]

List all assets asynchronously.

Arguments:

  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[Asset] - An asynchronous iterator over Asset objects.

get_asset_owner

def get_asset_owner(asset_id: str) -> Company

Get the company that owns an asset.

Arguments:

  • asset_id str - The unique identifier of the asset.

Returns:

  • Company - The Company object that owns the asset.

get_asset_owner_async

async def get_asset_owner_async(asset_id: str) -> Company

Get the company that owns an asset asynchronously.

Arguments:

  • asset_id str - The unique identifier of the asset.

Returns:

  • Company - The Company object that owns the asset.

search_assets

def search_assets(query: str,
                  scope: Literal["public", "company",
                                 "organization"] = "public",
                  company_id: str | None = None,
                  **extra_params: Any) -> PaginatedIterator[Asset]

Search for assets.

Arguments:

  • query str - The search query string. Asset names and addresses are searched for.
  • scope Literal["public", "company", "organization"] - The scope of the search. "public" is the default scope and searches all available assets in VELO. "organization" searches all private assets uploaded to the organization. If "company" is selected, company_id must also be provided.
  • company_id Optional[str] - The ID of the company to scope the search to. Required if scope is "company".
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[Asset] - A paginated iterator of assets matching the search criteria.

search_assets_async

async def search_assets_async(
        query: str,
        scope: Literal["public", "company", "organization"] = "public",
        company_id: str | None = None,
        **extra_params: Any) -> AsyncPaginatedIterator[Asset]

Search for assets asynchronously.

Arguments:

  • query str - The search query string. Asset names and addresses are searched for.
  • scope Literal["public", "company", "organization"] - The scope of the search. "public" is the default scope and searches all available assets in VELO. "organization" searches all private assets uploaded to the organization. If "company" is selected, company_id must also be provided.
  • company_id Optional[str] - The ID of the company to scope the search to. Required if scope is "company".
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[Asset] - An asynchronous paginated iterator of assets matching the search criteria.

velo_sdk.api.api_client

velo_sdk.api.types

Company Objects

class Company(BaseModel)

A company is an entity that has assets and identifying information.

Asset Objects

class Asset(BaseModel)

An asset represents aphysical asset that is subject to climate risk.

MarketIndex Objects

class MarketIndex(BaseModel)

A market index is a collection of companies.

ClimateScore Objects

class ClimateScore(BaseModel)

The cimate risk metrics that represent the likelihood of a company or asset to be impacted by climate risk.

ImpactScore Objects

class ImpactScore(BaseModel)

The impact risk metrics that represent the potential impact of a company or asset to be impacted by climate risk. These metrics represent an individual risk factor and its attribution to the total climate risk metrics.

CountryClimateScore Objects

class CountryClimateScore(ClimateScore)

Climate risk metrics aggregated for a country.

CountryImpactScore Objects

class CountryImpactScore(ImpactScore)

Impact risk metrics aggregated for a country.

AssetTypeClimateScore Objects

class AssetTypeClimateScore(ClimateScore)

Climate risk metrics aggregated for an asset type.

AssetTypeImpactScore Objects

class AssetTypeImpactScore(ImpactScore)

Impact risk metrics aggregated for an asset type.

AssetClimateScore Objects

class AssetClimateScore(ClimateScore)

Climate risk metrics for an asset.

AssetImpactScore Objects

class AssetImpactScore(BaseModel)

Impact risk metrics for an asset.

velo_sdk.api.markets

Markets Objects

class Markets()

search_indexes

def search_indexes(*,
                   name: Optional[str] = None,
                   **extra_params: Any) -> list[MarketIndex]

Search for market indexes by name.

Arguments:

  • name Optional[str] - The name of the market index to search for.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • list[MarketIndex] - A list of MarketIndex objects matching the search criteria.

search_indexes_async

async def search_indexes_async(*,
                               name: Optional[str] = None,
                               **extra_params: Any) -> list[MarketIndex]

Search for market indexes by name asynchronously.

Arguments:

  • name Optional[str] - The name of the market index to search for.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • list[MarketIndex] - A list of MarketIndex objects matching the search criteria.

list_indexes

def list_indexes() -> PaginatedIterator[MarketIndex]

List all market indexes.

Returns:

  • PaginatedIterator[MarketIndex] - An iterator over MarketIndex objects.

list_indexes_async

async def list_indexes_async() -> AsyncPaginatedIterator[MarketIndex]

List all market indexes asynchronously.

Returns:

  • AsyncPaginatedIterator[MarketIndex] - An asynchronous iterator over MarketIndex objects.

get_index

def get_index(index_id: str) -> MarketIndex

Get a market index by its unique ID.

Arguments:

  • index_id str - The unique identifier of the market index.

Returns:

  • MarketIndex - The MarketIndex object.

get_index_async

async def get_index_async(index_id: str) -> MarketIndex

Get a market index by its unique ID asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.

Returns:

  • MarketIndex - The MarketIndex object.

get_index_companies

def get_index_companies(index_id: str) -> PaginatedIterator[Company]

Get all companies in a market index.

Arguments:

  • index_id str - The unique identifier of the market index.

Returns:

  • PaginatedIterator[Company] - An iterator over Company objects in the index.

get_index_companies_async

async def get_index_companies_async(
        index_id: str) -> AsyncPaginatedIterator[Company]

Get all companies in a market index asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.

Returns:

  • AsyncPaginatedIterator[Company] - An asynchronous iterator over Company objects in the index.

get_index_climate_scores

def get_index_climate_scores(index_id: str, pathway: Pathway,
                             horizon: HorizonYear) -> ClimateScore

Get the climate scores for a market index.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • ClimateScore - The ClimateScore object for the market index.

get_index_climate_scores_async

async def get_index_climate_scores_async(index_id: str, pathway: Pathway,
                                         horizon: HorizonYear) -> ClimateScore

Get the climate scores for a market index asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • ClimateScore - The ClimateScore object for the market index.

get_index_impact_scores

def get_index_impact_scores(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[ImpactScore]

Get the impact scores for a market index.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[ImpactScore] - An iterator over ImpactScore objects for the market index.

get_index_impact_scores_async

async def get_index_impact_scores_async(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[ImpactScore]

Get the impact scores for a market index asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[ImpactScore] - An asynchronous iterator over ImpactScore objects for the market index.

list_index_asset_impact_scores

def list_index_asset_impact_scores(
        index_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetImpactScore]

Get the impact scores for all assets of a market index.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetImpactScore] - An iterator over AssetImpactScore objects for the index's assets.

list_index_asset_climate_scores

def list_index_asset_climate_scores(
        index_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> PaginatedIterator[AssetClimateScore]

Get the climate scores for all assets of a market index.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • PaginatedIterator[AssetClimateScore] - An iterator over AssetClimateScore objects for the index's assets.

list_index_asset_climate_scores_async

async def list_index_asset_climate_scores_async(
        index_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetClimateScore]

Get the climate scores for all assets of a market index asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetClimateScore] - An asynchronous iterator over AssetClimateScore objects for the index's assets.

list_index_asset_impact_scores_async

async def list_index_asset_impact_scores_async(
        index_id: str, pathway: Pathway, horizon: HorizonYear,
        **extra_params: Any) -> AsyncPaginatedIterator[AssetImpactScore]

Get the impact scores for all assets of a market index asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.
  • **extra_params Any - Additional parameters to pass to the API.

Returns:

  • AsyncPaginatedIterator[AssetImpactScore] - An asynchronous iterator over AssetImpactScore objects for the index's assets.

aggregate_index_asset_climate_scores_by_country

def aggregate_index_asset_climate_scores_by_country(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryClimateScore]

Get the climate scores for all assets in a market index aggregated by country.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryClimateScore] - An iterator over CountryClimateScore objects, aggregated by country.

aggregate_index_asset_climate_scores_by_country_async

async def aggregate_index_asset_climate_scores_by_country_async(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryClimateScore]

Get the climate scores for all assets in a market index aggregated by country asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryClimateScore] - An asynchronous iterator over CountryClimateScore objects, aggregated by country.

aggregate_index_asset_impact_scores_by_country

def aggregate_index_asset_impact_scores_by_country(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryImpactScore]

Get the impact scores for all assets in a market index aggregated by country.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryImpactScore] - An iterator over CountryImpactScore objects, aggregated by country.

aggregate_index_asset_impact_scores_by_country_async

async def aggregate_index_asset_impact_scores_by_country_async(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[CountryImpactScore]

Get the impact scores for all assets in a market index aggregated by country asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[CountryImpactScore] - An asynchronous iterator over CountryImpactScore objects, aggregated by country.

aggregate_index_asset_climate_scores_by_asset_type

def aggregate_index_asset_climate_scores_by_asset_type(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeClimateScore]

Get the climate scores for all assets in a market index aggregated by asset type.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeClimateScore] - An iterator over AssetTypeClimateScore objects, aggregated by asset type.

aggregate_index_asset_climate_scores_by_asset_type_async

async def aggregate_index_asset_climate_scores_by_asset_type_async(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeClimateScore]

Get the climate scores for all assets in a market index aggregated by asset type asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeClimateScore] - An asynchronous iterator over AssetTypeClimateScore objects, aggregated by asset type.

aggregate_index_asset_impact_scores_by_asset_type

def aggregate_index_asset_impact_scores_by_asset_type(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeImpactScore]

Get the impact scores for all assets in a market index aggregated by asset type.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeImpactScore] - An iterator over AssetTypeImpactScore objects, aggregated by asset type.

aggregate_index_asset_impact_scores_by_asset_type_async

async def aggregate_index_asset_impact_scores_by_asset_type_async(
        index_id: str, pathway: Pathway,
        horizon: HorizonYear) -> StaticListIterator[AssetTypeImpactScore]

Get the impact scores for all assets in a market index aggregated by asset type asynchronously.

Arguments:

  • index_id str - The unique identifier of the market index.
  • pathway Pathway - Climate scenario pathway powered by Climate Earth Digital Twin.
  • horizon HorizonYear - Climatology year representing a decadal period.

Returns:

  • StaticListIterator[AssetTypeImpactScore] - An asynchronous iterator over AssetTypeImpactScore objects, aggregated by asset type.