Skip to content
Open
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
28 changes: 20 additions & 8 deletions blockapi/v2/api/debank.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

from blockapi.utils.address import make_checksum_address
from blockapi.utils.datetime import parse_dt
from blockapi.utils.num import decimals_to_raw, to_decimal
from blockapi.utils.num import decimals_to_raw
from blockapi.v2.api.debank_maps import (
COINGECKO_IDS_BY_CONTRACTS,
DEBANK_APP_CHAIN_MAP,
Expand Down Expand Up @@ -42,6 +42,7 @@
DebankModelAppPortfolioItem,
DebankModelApp,
DebankModelPredictionDetail,
DebankDepositToken,
)

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -609,19 +610,19 @@ def parse(self, response: list) -> list[DebankApp]:

apps = []
for item in response:
app = self._parse_app(item)
try:
app = self._parse_app(item)
except Exception as e:
logger.error(f'Failed to parse app: {e}')
continue
if app:
apps.append(app)

return apps

def _parse_app(self, raw_app: dict) -> Optional[DebankApp]:
"""Parse a single app from the response."""
try:
model = DebankModelApp(**raw_app)
except Exception as e:
logger.error(f'Failed to parse app: {e}')
return None
model = DebankModelApp(**raw_app)

deposits = []
predictions = []
Expand Down Expand Up @@ -679,17 +680,28 @@ def _parse_deposit(
self, item: DebankModelAppPortfolioItem, chain: Optional[Blockchain]
) -> Optional[DebankAppDeposit]:
"""Parse a deposit/common type portfolio item."""
parsed_tokens = [
self._parse_token(t.model_dump()) for t in item.asset_token_list or []
]

return DebankAppDeposit.from_api(
name=item.name,
asset_usd_value=item.stats.asset_usd_value,
debt_usd_value=item.stats.debt_usd_value,
net_usd_value=item.stats.net_usd_value,
tokens=item.asset_token_list,
tokens=[t for t in parsed_tokens if t is not None],
chain=chain,
position_index=item.position_index,
update_at=item.update_at,
)

def _parse_token(self, raw_token: dict) -> Optional[DebankDepositToken]:
try:
return DebankDepositToken.from_api(**raw_token)
except Exception as e:
logger.error(f'Failed to parse deposit token: {e}')
return None


class DebankApi(CustomizableBlockchainApi, BalanceMixin, IPortfolio):
"""
Expand Down
37 changes: 35 additions & 2 deletions blockapi/v2/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1197,7 +1197,7 @@ class DebankModelPredictionDetail(BaseModel):
event_end_at: Optional[float] = None


class DebankDepositToken(BaseModel):
class DebankModelDepositToken(BaseModel):
"""Token within deposit/common type portfolio items."""

id: str
Expand All @@ -1218,7 +1218,7 @@ class DebankModelAppPortfolioItem(BaseModel):
detail: dict
position_index: str
asset_dict: Optional[dict] = None
asset_token_list: list[DebankDepositToken] = Field(default_factory=list)
asset_token_list: list[DebankModelDepositToken] = Field(default_factory=list)
update_at: Optional[float] = None
proxy_detail: Optional[dict] = None

Expand Down Expand Up @@ -1281,6 +1281,39 @@ def from_api(
)


@attr.s(auto_attribs=True, slots=True, frozen=True)
class DebankDepositToken:
id: str
symbol: str
name: str
amount: Decimal
app_id: str
price: Decimal
logo_url: Optional[str]

@classmethod
def from_api(
cls,
*,
id: str,
symbol: str,
name: str,
amount: Union[str, float, int],
app_id: str,
price: Union[str, float, int],
logo_url: Optional[str] = None,
) -> 'DebankDepositToken':
return cls(
id=id,
symbol=symbol,
name=name,
amount=to_decimal(amount),
app_id=app_id,
price=to_decimal(price),
logo_url=logo_url,
)


@attr.s(auto_attribs=True, slots=True, frozen=True)
class DebankAppDeposit:
"""Represents a deposit/holding within a Debank App (e.g., Polymarket cash deposit)."""
Expand Down
Loading