Dev

deltadefi/api_resources/api_config.py

@@ -0,0 +1,11 @@
+from typing import TypedDict, Literal
+from typing_extensions import NotRequired
+
+class AppWalletKeyType(str):
+    pass # TODO: import AppWalletKeyType
+
+class ApiConfig(TypedDict, total=False):
+    network: NotRequired[Literal['preprod', 'mainnet']]
+    signingKey: NotRequired[AppWalletKeyType]
+    jwt: NotRequired[str]
+    apiKey: NotRequired[str]

deltadefi/api_resources/auth.py

@@ -0,0 +1,11 @@
+from typing import TypedDict
+from typing_extensions import NotRequired
+
+class AuthHeaders(TypedDict, total=False):
+    jwt: str
+    apiKey: str
+
+class ApiHeaders(TypedDict):
+    'Content-Type': str
+    Authorization: NotRequired[str]
+    'X-API-KEY': NotRequired[str]

deltadefi/api_resources/conversion.py

@@ -0,0 +1,33 @@
+from typing import TypedDict, List
+from deltadefi.requests.params import InputUtxos
+
+class Asset(TypedDict): 
+    pass # TODO: import Asset
+
+class UTxO(TypedDict): # TODO: import UTxO
+    pass
+
+class TxInParameter(TypedDict): # TODO: import TxInParameter
+    txHash: str
+    txIndex: int
+    amount: List[Asset]
+    address: str
+
+def convert_utxo(utxo: UTxO) -> InputUtxos:
+    return {
+        'tx_hash': utxo['input']['txHash'],
+        'tx_id': str(utxo['input']['outputIndex']),
+        'amount': utxo['output']['amount'],
+        'address': utxo['output']['address']
+    }
+
+def convert_utxos(utxos: List[UTxO]) -> List[InputUtxos]:
+    return [convert_utxo(utxo) for utxo in utxos]
+
+def convert_tx_in_parameter(tx_in: TxInParameter) -> InputUtxos:
+    return {
+        'tx_hash': tx_in['txHash'],
+        'tx_id': str(tx_in['txIndex']),
+        'amount': tx_in.get('amount', []),
+        'address': tx_in.get('address', '')
+    }

deltadefi/api_resources/validation.py

@@ -0,0 +1,36 @@
+from typing import TypedDict, List
+
+class Asset(TypedDict):
+    pass # TODO: import Asset
+
+"""
+ * DeltaDeFiOrderInfo is a type that represents the information of a DeltaDeFi order.
+ * @property {Asset[]} assetsToPay - The assets that are to be paid from orders in current transaction.
+ * @property {Asset[]} assetsToReturn - The assets that are to be received from orders in current transaction.
+ * @property {string} txFee - The transaction fee.
+ * @property {string} tradingFee - The trading fee.
+"""
+
+class DeltaDeFiOrderInfo(TypedDict):
+    assetsToPay: List[Asset]
+    assetsToReturn: List[Asset]
+    txFee: str
+    tradingFee: str
+
+"""
+ * DeltaDeFiTxInfo is a type that represents the information of a DeltaDeFi transaction.
+ * @property {Asset[]} accountInput - The assets that are input from the account.
+ * @property {Asset[]} accountOutput - The assets that are output to the account.
+ * @property {Asset[]} dexInput - The assets that are input from the DEX.
+ * @property {Asset[]} dexOutput - The assets that are output to the DEX.
+ * @property {string} txFee - The transaction fee.
+ * @property {string} tradingFee - The trading fee.
+"""
+
+class DeltaDeFiTxInfo(TypedDict):
+    accountInput: List[Asset]
+    accountOutput: List[Asset]
+    dexInput: List[DeltaDeFiOrderInfo]
+    dexOutput: List[DeltaDeFiOrderInfo]
+    txFee: str
+    tradingFee: str

deltadefi/api_resources/value.py

@@ -0,0 +1,151 @@
+from typing import Dict, List, Tuple
+
+class Asset: # TODO: import Asset
+    unit: str
+    quantity: int
+
+class Value:
+    value: Dict[str, int]
+
+    def __init__(self) -> None:
+        self.value = {}
+
+    """
+    * Add an asset to the Value class's value record. If an asset with the same unit already exists in the value record, the quantity of the
+    * existing asset will be increased by the quantity of the new asset. If no such asset exists, the new asset will be added to the value record.
+    * Implementation:
+    * 1. Check if the unit of the asset already exists in the value record.
+    * 2. If the unit exists, add the new quantity to the existing quantity.
+    * 3. If the unit does not exist, add the unti to the object.
+    * 4. Return the Value class instance.
+    * @param asset
+    * @returns this
+    """
+    def add_asset(self, asset: Asset) -> 'Value':
+        quantity = asset.quantity
+        unit = asset.unit
+
+        if unit in self.value:
+            self.value[unit] += quantity
+        else:
+            self.value[unit] = quantity
+        return self
+
+    """
+     * Add an array of assets to the Value class's value record. If an asset with the same unit already exists in the value record, the quantity of the
+     * existing asset will be increased by the quantity of the new asset. If no such asset exists, the new assets under the array of assets will be added to the value record.
+     * Implementation:
+     * 1. Iterate over each asset in the 'assets' array.
+     * 2. For each asset, check if the unit of the asset already exists in the value record.
+     * 3. If the unit exists, add the new quantity to the existing quantity.
+     * 4. If the unit does not exist, add the unti to the object.
+     * 5. Return the Value class instance.
+     * @param assets
+     * @returns this
+    """
+    def add_assets(self, assets: List[Asset]) -> 'Value':
+        for asset in assets:
+            self.add_asset(asset)
+        return self
+
+    """
+     * Substract an asset from the Value class's value record. If an asset with the same unit already exists in the value record, the quantity of the
+     * existing asset will be decreased by the quantity of the new asset. If no such asset exists, an error message should be printed.
+     * Implementation:
+     * 1. Check if the unit of the asset already exists in the value record.
+     * 2. If the unit exists, subtract the new quantity from the existing quantity.
+     * 3. If the unit does not exist, print an error message.
+     * @param asset
+     * @returns this
+    """
+    def negate_asset(self, asset: Asset) -> 'Value':
+        unit = asset.unit
+        quantity = asset.quantity
+
+        current_quantity = self.value.get(unit, 0)
+        new_quantity = current_quantity - quantity
+
+        if new_quantity == 0:
+            del self.value[unit]
+        else:
+            self.value[unit] = new_quantity
+        return self
+
+    """
+     * Subtract an array of assets from the Value class's value record. If an asset with the same unit already exists in the value record, the quantity of the
+     * existing asset will be decreased by the quantity of the new asset. If no such asset exists, an error message should be printed.
+     * @param assets
+     * @returns this
+    """
+    def negate_assets(self, assets: List[Asset]) -> 'Value':
+        for asset in assets:
+            self.negate_asset(asset)
+        return self
+
+    # """
+    #  * Get the quantity of asset object per unit
+    #  * @param unit
+    #  * @returns
+    # """
+    # def get(self, unit: str) -> int:
+    #     return self.value.get(unit, 0)
+
+    """
+     * Get all assets (return Record of Asset[])
+     * @param
+     * @returns Record<string, Asset[]>
+    """
+    def units(self) -> Dict[str, List[Tuple[str, int]]]:
+        result = {}
+        for unit, quantity in self.value.items():
+            if unit not in result:
+                result[unit] = []
+            result[unit].append((unit, quantity))
+        return result
+
+    """
+     * Check if the value is greater than or equal to an inputted value
+     * @param unit - The unit to compare (e.g., "ADA")
+     * @param other - The value to compare against
+     * @returns boolean
+    """
+    def geq(self, unit: str, other: 'Value') -> bool:
+        if unit not in self.value or unit not in other.value:
+            return False
+        return self.value[unit] >= other.value[unit]
+
+    """
+     * Check if the value is less than or equal to an inputted value
+     * @param unit - The unit to compare (e.g., "ADA")
+     * @param other - The value to compare against
+     * @returns boolean
+    """
+    def leq(self, unit: str, other: 'Value') -> bool:
+        if unit not in self.value or unit not in other.value:
+            return False
+        return self.value[unit] <= other.value[unit]
+
+    """
+     * Check if the value is empty
+     * @param
+     * @returns boolean
+    """
+    def is_empty(self) -> bool:
+        return not self.value
+
+    """
+     * Merge the given values
+     * @param values
+     * @returns this
+    """
+    def merge(self, values: 'Value' | List['Value']) -> 'Value':
+        if isinstance(values, list):
+            values_list = values
+        else:
+            values_list = [values]
+
+        for other in values_list:
+            for unit, quantity in other.value.items():
+                self.value[unit] = self.value.get(unit, 0) + quantity
+
+        return self

deltadefi/api_resources/wallet.py

@@ -0,0 +1,4 @@
+class DeFiWallet:
+    def __init__(self, signing_key, network_id):
+        self.signing_key = signing_key
+        self.network_id = network_id

deltadefi/clients/__init__.py

@@ -0,0 +1,34 @@
+from deltadefi.api_resources.wallet import DeFiWallet
+from deltadefi.clients.accounts import Accounts
+from deltadefi.clients.orders import Orders
+from deltadefi.clients.markets import Markets
+from deltadefi.api_resources.api_config import ApiConfig
+
+class ApiClient:
+
+    def __init__(self, config: ApiConfig, provided_base_url: str = None):
+        if config.get('network') == 'mainnet': 
+            self.network_id = 1
+            self.base_url = 'https://api-dev.deltadefi.io'
+        elif config.get('network') == 'preprod':
+            self.base_url = 'https://api-dev.deltadefi.io'  # TODO: input production link once available
+            self.network_id = 0 
+        else:
+            raise ConnectionRefusedError
+
+        self.headers = {
+            'Content-Type': 'application/json'
+        }
+        if config.get('jwt'):
+            self.headers['Authorization'] = config['jwt']
+        if config.get('apiKey'):
+            self.headers['X-API-KEY'] = config['apiKey']
+        if config.get('signingKey'): # TODO: import DefiWallet from mesh
+            self.defiWallet = DeFiWallet(self)
+
+        if provided_base_url:
+            self.base_url = provided_base_url
+
+        self.accounts = Accounts(self)
+        self.orders = Orders(self)
+        self.markets = Markets(self)