Feature portfolio PNL

.env.example

@@ -13,6 +13,9 @@ SOLANA_RPC_URL=https://api.mainnet-beta.solana.com
 # Alchemy Configuration
 ALCHEMY_API_KEY=
 
+# HELIUS API for Solana
+HELIUS_API_KEY=
+
 # Cookie.fun Configuration
 COOKIE_FUN_API_KEY=
 

alphaswarm/config.py

@@ -230,9 +230,14 @@ def _filter_networks(self) -> None:
 
     def get_supported_networks(self) -> list:
         """Get list of supported networks for current environment"""
-        if self._network_env == "all":
-            return self._config["chain_config"].keys()
-        return self._config["network_environments"].get(self._network_env, [])
+        network_env = self._config["network_environments"]
+        if self._network_env != "all":
+            return network_env.get(self._network_env, [])
+
+        result = set()
+        for networks in network_env.values():
+            result.update(set(networks))
+        return list(result)
 
     def get(self, key_path: str, default: Any = None) -> Any:
         """Get configuration value using dot notation"""

alphaswarm/services/alchemy/alchemy_client.py

@@ -3,6 +3,7 @@
 import logging
 import os
 import time
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from decimal import Decimal
 from typing import Annotated, Dict, Final, List, Optional
@@ -34,6 +35,17 @@ class Metadata(BaseModel):
     block_timestamp: Annotated[str, Field(alias="blockTimestamp")]
 
 
+@dataclass
+class RawContract:
+    address: str
+    value: int
+    decimal: int
+
+    @field_validator("value", "decimal", mode="before")
+    def convert_hex_to_int(cls, value: str) -> int:
+        return int(value, 16)
+
+
 class Transfer(BaseModel):
     """Represents a token transfer transaction.
 
@@ -57,6 +69,7 @@ class Transfer(BaseModel):
     from_address: Annotated[str, Field(validation_alias="from")]
     to_address: Annotated[str, Field(validation_alias="to")]
     value: Annotated[Decimal, Field(default=Decimal(0))]
+    raw_contract: Annotated[RawContract, Field(alias="rawContract")]
     metadata: Metadata
     asset: str = "UNKNOWN"
     category: str = "UNKNOWN"

alphaswarm/services/chains/solana/solana_client.py

@@ -8,12 +8,13 @@
 from alphaswarm.services.chains.solana.jupiter_client import JupiterClient
 from pydantic import BaseModel, Field
 from solana.rpc import api
+from solana.rpc.commitment import Finalized
 from solana.rpc.types import TokenAccountOpts
 from solders.account_decoder import ParsedAccount
 from solders.keypair import Keypair
 from solders.message import to_bytes_versioned
 from solders.pubkey import Pubkey
-from solders.rpc.responses import SendTransactionResp
+from solders.rpc.responses import RpcConfirmedTransactionStatusWithSignature, SendTransactionResp
 from solders.signature import Signature
 from solders.transaction import VersionedTransaction
 from solders.transaction_status import TransactionConfirmationStatus
@@ -161,3 +162,12 @@ def _wait_for_confirmation(self, signature: Signature) -> None:
         raise RuntimeError(
             f"Failed to get confirmation for transaction '{str(signature)}' for {initial_timeout} seconds. Last status is {status}"
         )
+
+    def get_signatures_for_address(
+        self, wallet_address: Pubkey, limit: int = 1000, before: Optional[Signature] = None
+    ) -> List[RpcConfirmedTransactionStatusWithSignature]:
+
+        result = self._client.get_signatures_for_address(
+            wallet_address, commitment=Finalized, limit=limit, before=before
+        )
+        return result.value

alphaswarm/services/helius/__init__.py

@@ -0,0 +1,2 @@
+from .helius_client import HeliusClient
+from .data import EnhancedTransaction, SignatureResult, TokenTransfer

alphaswarm/services/helius/data.py

@@ -0,0 +1,188 @@
+from decimal import Decimal
+from enum import StrEnum
+from typing import Annotated, Dict, List, Optional
+
+from pydantic import Field
+from pydantic.dataclasses import dataclass
+
+
+class ConfirmationStatus(StrEnum):
+    CONFIRMED = "confirmed"
+    FINALIZED = "finalized"
+    PROCESSED = "processed"
+
+
+@dataclass
+class SignatureResult:
+    signature: str
+    slot: int
+    err: Annotated[Optional[Dict], Field(default=None)]
+    memo: Annotated[Optional[str], Field(default=None)]
+    block_time: Annotated[Optional[int], Field(default=None, alias="blockTime")]
+    confirmation_status: Annotated[Optional[ConfirmationStatus], Field(default=None, alias="confirmationStatus")]
+
+
+@dataclass
+class NativeTransfer:
+    from_user_account: Annotated[str, Field(alias="fromUserAccount")]
+    to_user_account: Annotated[str, Field(alias="toUserAccount")]
+    amount: Annotated[int, Field(description="The amount fo sol sent (in lamports)")]
+
+
+@dataclass
+class TokenTransfer:
+    from_user_account: Annotated[str, Field(alias="fromUserAccount")]
+    to_user_account: Annotated[str, Field(alias="toUserAccount")]
+    from_token_account: Annotated[str, Field(alias="fromTokenAccount")]
+    to_token_account: Annotated[str, Field(alias="toTokenAccount")]
+    token_amount: Annotated[Decimal, Field(alias="tokenAmount", description="The number of tokens sent.")]
+    mint: str
+
+
+@dataclass
+class RawTokenAmount:
+    token_amount: Annotated[Decimal, Field(alias="tokenAmount")]
+    decimals: int
+
+
+@dataclass
+class TokenBalanceChange:
+    user_account: Annotated[str, Field(alias="userAccount")]
+    token_account: Annotated[str, Field(alias="tokenAccount")]
+    mint: str
+    raw_token_amount: Annotated[RawTokenAmount, Field(alias="rawTokenAmount")]
+
+
+@dataclass
+class InnerInstruction:
+    data: str
+    program_id: Annotated[str, Field(alias="programId")]
+    accounts: List[str]
+
+
+@dataclass
+class Instruction:
+    data: str
+    program_id: Annotated[str, Field(alias="programId")]
+    accounts: List[str]
+    inner_instructions: Annotated[List[InnerInstruction], Field(alias="innerInstructions")]
+
+
+@dataclass
+class TransactionError:
+    error: str
+
+
+@dataclass
+class Nft:
+    mint: str
+    token_standard: Annotated[str, Field(alias="tokenStandard")]
+
+
+@dataclass
+class NftEvent:
+    description: str
+    type: str
+    source: str
+    amount: Annotated[int, Field(description="The amount of the NFT transaction (in lamports)")]
+    fee: int
+    fee_payer: Annotated[str, Field(alias="feePayer")]
+    signature: str
+    slot: int
+    timestamp: int
+    sale_type: Annotated[str, Field(alias="saleType")]
+    buyer: str
+    seller: str
+    staker: str
+    ntfs: List[Nft]
+
+
+@dataclass
+class AccountData:
+    account: str
+    native_balance_change: Annotated[
+        Decimal, Field(alias="nativeBalanceChange", description="Native (SOL) balance change of the account.")
+    ]
+    token_balance_changes: Annotated[List[TokenBalanceChange], Field(alias="tokenBalanceChanges")]
+
+
+@dataclass
+class NativeAmount:
+    account: str
+    amount: Annotated[str, Field(description="The amount of the balance change")]
+
+
+@dataclass
+class ProgramInfo:
+    source: str
+    account: str
+    program_name: Annotated[str, Field(alias="programName")]
+    instruction_name: Annotated[str, Field(alias="instructionName")]
+
+
+@dataclass
+class InnerSwap:
+    program_info: ProgramInfo
+    token_inputs: Annotated[List[TokenTransfer], Field(alias="tokenInputs")]
+    token_outputs: Annotated[List[TokenTransfer], Field(alias="tokenOutputs")]
+    token_fees: Annotated[List[TokenTransfer], Field(alias="tokenFees")]
+    native_fees: Annotated[List[NativeTransfer], Field(alias="nativeFees")]
+
+
+@dataclass
+class SwapEvent:
+    native_input: Annotated[NativeAmount, Field(alias="nativeAmount")]
+    native_output: Annotated[NativeAmount, Field(alias="nativeAmount")]
+    token_inputs: Annotated[List[TokenBalanceChange], Field(alias="tokenInputs")]
+    token_outputs: Annotated[List[TokenBalanceChange], Field(alias="tokenOutputs")]
+    tokens_fees: Annotated[List[TokenBalanceChange], Field(alias="tokensFees")]
+    native_fees: Annotated[List[NativeAmount], Field(alias="nativeFees")]
+    inner_swaps: Annotated[List[InnerSwap], Field(alias="innerSwaps")]
+
+
+@dataclass
+class Compressed:
+    type: str
+    tree_id: Annotated[str, Field(alias="treeId")]
+    asset_id: Annotated[str, Field(alias="assetId")]
+    leaf_index: Annotated[int, Field(alias="leafIndex")]
+    instruction_index: Annotated[int, Field(alias="instructionIndex")]
+    inner_instruction_index: Annotated[int, Field(alias="innerInstructionIndex")]
+    new_leaf_owner: Annotated[str, Field(alias="newLeafOwner")]
+    old_leaf_owner: Annotated[str, Field(alias="oldLeafOwner")]
+
+
+@dataclass
+class Authority:
+    account: str
+    from_: Annotated[str, Field(alias="from")]
+    to: str
+    instruction_index: Annotated[int, Field(alias="instructionIndex")]
+    inner_instruction_index: Annotated[int, Field(alias="innerInstructionIndex")]
+
+
+@dataclass
+class TransactionEvent:
+    nft: NftEvent
+    swap: SwapEvent
+    compressed: Compressed
+    distributed_compression_rewards: Annotated[int, Field(alias="distributedCompressionRewards")]
+
+
+@dataclass
+class EnhancedTransaction:
+    description: str
+    type: str
+    source: str
+    fee: int
+    fee_payer: Annotated[str, Field(alias="feePayer")]
+    signature: str
+    slot: int
+    timestamp: int
+    native_transfers: Annotated[List[NativeTransfer], Field(alias="nativeTransfers")]
+    token_transfers: Annotated[List[TokenTransfer], Field(alias="tokenTransfers")]
+    account_data: Annotated[List[AccountData], Field(alias="accountData")]
+    instructions: List[Instruction]
+    # Disabled for simplicity for now.
+    # transaction_error: Annotated[Optional[TransactionError], Field(alias="transactionError", default=None)]
+    # events: TransactionEvent

alphaswarm/services/helius/helius_client.py

@@ -0,0 +1,47 @@
+import os
+from typing import Dict, Final, List, Self, Sequence
+
+import requests
+from alphaswarm.services import ApiException
+
+from .data import EnhancedTransaction, SignatureResult
+
+
+class HeliusClient:
+    BASE_RPC_URL: Final[str] = "https://mainnet.helius-rpc.com"
+    BASE_TRANSACTION_URL: Final[str] = "https://api.helius.xyz"
+
+    def __init__(self, api_key: str) -> None:
+        self._api_key = api_key
+
+    def _make_request(self, url: str, data: Dict) -> Dict:
+        params = {"api-key": self._api_key}
+        response = requests.post(url=url, json=data, params=params)
+        if response.status_code >= 400:
+            raise ApiException(response)
+        return response.json()
+
+    def get_signatures_for_address(self, wallet_address: str) -> List[SignatureResult]:
+        body = {
+            "id": 1,
+            "jsonrpc": "2.0",
+            "method": "getSignaturesForAddress",
+            "params": [
+                wallet_address,
+            ],
+        }
+
+        response = self._make_request(self.BASE_RPC_URL, body)
+        return [SignatureResult(**item) for item in response["result"]]
+
+    def get_transactions(self, signatures: Sequence[str]) -> List[EnhancedTransaction]:
+        if len(signatures) > 100:
+            raise ValueError("Can only get 100 transactions at a time")
+
+        data = {"transactions": signatures}
+        response = self._make_request(f"{self.BASE_TRANSACTION_URL}/v0/transactions", data)
+        return [EnhancedTransaction(**item) for item in response]
+
+    @classmethod
+    def from_env(cls) -> Self:
+        return cls(api_key=os.environ["HELIUS_API_KEY"])

alphaswarm/services/portfolio/__init__.py

@@ -1 +1,5 @@
 from .portfolio import Portfolio
+from .portfolio_base import PortfolioBase, PortfolioSwap
+from .portfolio_evm import PortfolioEvm
+from .portfolio_pnl import PortfolioPNL, PortfolioPNLDetail, PnlMode
+from .portfolio_solana import PortfolioSolana