Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion src/nansen/resources/points.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@


class Points(SyncAPIResource):
"""Points resource for the Nansen points leaderboard (no auth required)."""

def leaderboard(
self,
*,
Expand All @@ -19,6 +21,9 @@ def leaderboard(

This endpoint uses GET, requires no authentication,
and hits a different base URL (app.nansen.ai).

Args:
tier: Filter by tier name.
"""
params = {}
if not isinstance(tier, _NotGiven):
Expand All @@ -36,12 +41,21 @@ def leaderboard(


class AsyncPoints(AsyncAPIResource):
"""Points resource for the Nansen points leaderboard (no auth required, async)."""

async def leaderboard(
self,
*,
tier: str | NotGiven = NOT_GIVEN,
) -> APIResponse[list[PointsLeaderboardEntry]]:
"""Fetch the points leaderboard (async)."""
"""Fetch the points leaderboard.

This endpoint uses GET, requires no authentication,
and hits a different base URL (app.nansen.ai).

Args:
tier: Filter by tier name.
"""
params = {}
if not isinstance(tier, _NotGiven):
params["tier"] = tier
Expand Down
14 changes: 14 additions & 0 deletions src/nansen/resources/portfolio.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,18 @@


class Portfolio(SyncAPIResource):
"""Portfolio resource for wallet-level analytics."""

def defi_holdings(
self,
*,
wallet_address: str,
) -> APIResponse[DefiHoldingsResponse]:
"""Get DeFi holdings for a wallet address.

Args:
wallet_address: The wallet address to look up.
"""
return self._post(
"/portfolio/defi-holdings",
body={"wallet_address": wallet_address},
Expand All @@ -19,11 +26,18 @@ def defi_holdings(


class AsyncPortfolio(AsyncAPIResource):
"""Portfolio resource for wallet-level analytics (async)."""

async def defi_holdings(
self,
*,
wallet_address: str,
) -> APIResponse[DefiHoldingsResponse]:
"""Get DeFi holdings for a wallet address.

Args:
wallet_address: The wallet address to look up.
"""
return await self._post(
"/portfolio/defi-holdings",
body={"wallet_address": wallet_address},
Expand Down
135 changes: 133 additions & 2 deletions src/nansen/resources/profiler/address.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@


class Address(SyncAPIResource):
"""Address sub-resource for wallet-level profiler endpoints."""

def current_balance(
self,
*,
Expand All @@ -28,6 +30,17 @@ def current_balance(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> SyncPage[ProfilerBalanceItem]:
"""Get current token balances for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
hide_spam_token: If ``True``, exclude tokens flagged as spam.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return self._post_page(
"/profiler/address/current-balance",
body={
Expand All @@ -53,6 +66,17 @@ def historical_balances(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> SyncPage[HistoricalBalanceItem]:
"""Get historical token balances for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return self._post_page(
"/profiler/address/historical-balances",
body={
Expand All @@ -78,6 +102,17 @@ def transactions(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> SyncPage[TransactionItem]:
"""Get transaction history for an address.

Args:
address: Wallet address to look up.
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
hide_spam_token: If ``True``, exclude transactions involving spam tokens.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return self._post_page(
"/profiler/address/transactions",
body={
Expand Down Expand Up @@ -105,6 +140,19 @@ def counterparties(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> SyncPage[CounterpartyItem]:
"""Get counterparty data for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
source_input: Source input type for the query.
group_by: Field to group counterparties by.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return self._post_page(
"/profiler/address/counterparties",
body={
Expand All @@ -129,6 +177,14 @@ def related_wallets(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> SyncPage[RelatedWalletItem]:
"""Get wallets related to an address.

Args:
address: Wallet address to look up.
chain: Chain identifier (e.g. ``"ethereum"``).
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return self._post_page(
"/profiler/address/related-wallets",
body={
Expand All @@ -149,10 +205,17 @@ def labels(
label: str | NotGiven = NOT_GIVEN,
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
) -> APIResponse[list[AddressLabelItem]]:
"""Fetch address labels (beta endpoint).
"""Get labels for an address (beta endpoint).

Note: This endpoint uses ``/api/beta/`` and has a different
This endpoint uses the ``/api/beta/`` base path and has a different
request structure than standard v1 endpoints.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
address: Wallet address to look up.
entity: Filter by entity name.
label: Filter by label name.
pagination: Pagination options (``page``, ``per_page``).
"""

parameters: dict[str, Any] = {"chain": chain, "address": address}
Expand Down Expand Up @@ -182,6 +245,8 @@ def labels(


class AsyncAddress(AsyncAPIResource):
"""Address sub-resource for wallet-level profiler endpoints (async)."""

async def current_balance(
self,
*,
Expand All @@ -193,6 +258,17 @@ async def current_balance(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> AsyncPage[ProfilerBalanceItem]:
"""Get current token balances for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
hide_spam_token: If ``True``, exclude tokens flagged as spam.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return await self._post_page(
"/profiler/address/current-balance",
body={
Expand All @@ -218,6 +294,17 @@ async def historical_balances(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> AsyncPage[HistoricalBalanceItem]:
"""Get historical token balances for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return await self._post_page(
"/profiler/address/historical-balances",
body={
Expand All @@ -243,6 +330,17 @@ async def transactions(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> AsyncPage[TransactionItem]:
"""Get transaction history for an address.

Args:
address: Wallet address to look up.
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
hide_spam_token: If ``True``, exclude transactions involving spam tokens.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return await self._post_page(
"/profiler/address/transactions",
body={
Expand Down Expand Up @@ -270,6 +368,19 @@ async def counterparties(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> AsyncPage[CounterpartyItem]:
"""Get counterparty data for an address or entity.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
date: Date range with ``"from"`` and ``"to"`` keys.
address: Wallet address to look up.
entity_name: Entity name to look up (alternative to ``address``).
source_input: Source input type for the query.
group_by: Field to group counterparties by.
filters: Field-level filters to narrow results.
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return await self._post_page(
"/profiler/address/counterparties",
body={
Expand All @@ -294,6 +405,14 @@ async def related_wallets(
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
order_by: list[dict[str, str]] | NotGiven = NOT_GIVEN,
) -> AsyncPage[RelatedWalletItem]:
"""Get wallets related to an address.

Args:
address: Wallet address to look up.
chain: Chain identifier (e.g. ``"ethereum"``).
pagination: Pagination options (``page``, ``per_page``).
order_by: List of ordering directives.
"""
return await self._post_page(
"/profiler/address/related-wallets",
body={
Expand All @@ -314,6 +433,18 @@ async def labels(
label: str | NotGiven = NOT_GIVEN,
pagination: dict[str, Any] | NotGiven = NOT_GIVEN,
) -> APIResponse[list[AddressLabelItem]]:
"""Get labels for an address (beta endpoint).

This endpoint uses the ``/api/beta/`` base path and has a different
request structure than standard v1 endpoints.

Args:
chain: Chain identifier (e.g. ``"ethereum"``).
address: Wallet address to look up.
entity: Filter by entity name.
label: Filter by label name.
pagination: Pagination options (``page``, ``per_page``).
"""

parameters: dict[str, Any] = {"chain": chain, "address": address}
if not isinstance(entity, NotGiven):
Expand Down
Loading