1
0

Example CLI Application
Some checks failed
Verify Python project can be installed, loaded and have version checked / Test (push) Waiting to run
Python Ruff Code Quality / ruff (push) Successful in 23s
Run Python tests (through Pytest) / Test (push) Has been cancelled

This commit is contained in:
Jon Michael Aanes 2024-11-29 00:45:31 +01:00
parent 705b8153dc
commit ba988fed6c
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
3 changed files with 168 additions and 1 deletions

View File

@ -39,6 +39,49 @@ Uses their [API](https://www.nordnet.dk/externalapi/docs/api). Thanks to
utilities](https://github.com/helmstedt/nordnet-utilities), which helped with
implementing this functionality. Exposes the same data as the home page.
## Example Application
The library ships with an example application in the form of a CLI program that
can display a tree representation of assets in known depositories:
```bash
python -m fin_depo
```
Example output:
```
Aggregated
Kraken
BTC 0.00871009
EUR 200.5
Kucoin
Kucoin trade
BTC 0.00169297
USDT 20
Kucoin main
BTC 0
USDT 40
```
## Usage
Using the Kraken API as an example:
```python
depo_fetcher = fin_depo.defi_kraken.KrakenDepoFetcher(
KRAKEN_KEY, KRAKEN_SECRET,
)
depo = depo_fetcher.get_depo()
depo.assets()
> [BTC, USDT]
depo.get_amount_of_asset(BTC)
> 0.1
```
## Future extension
- [ ] Investment Bank: Saxo Bank OpenAPI

122
fin_depo/__main__.py Normal file
View File

@ -0,0 +1,122 @@
"""# Example Depository CLI Tool.
This example script demonstrates how to initialize the depository fetchers, and
how to access the downloaded data.
"""
import sys
import requests
from secret_loader import SecretLoader
import fin_depo
secret_loader = SecretLoader(ENV_KEY_PREFIX='CF_FD')
DISABLE_NORDNET = True
def setup_aggregate_depos() -> fin_depo.static.AggregateDepoFetcher:
"""Initializes a composite depository fetcher.
The secret handling implemented here is robust, but possibly overkill for
your application.
"""
session = requests.Session()
depo_fetchers = []
if pbc_address := secret_loader.load('PBC_ACCOUNT_ADDRESS'):
depo_fetchers.append(
fin_depo.defi_partisia_blockchain.PartisiaBlockchainAccountDepoFetcher(
session,
pbc_address,
),
)
del pbc_address
if (
nordnet_username := secret_loader.load('NORDNET_USERNAME')
and not DISABLE_NORDNET
):
depo_fetchers.append(
fin_depo.investbank_nordnet.NordnetDepoFetcher(
session,
nordnet_username,
secret_loader.load_or_fail('NORDNET_PASSWORD'),
),
)
del nordnet_username
if kraken_key := secret_loader.load('KRAKEN_KEY'):
depo_fetchers.append(
fin_depo.defi_kraken.KrakenDepoFetcher(
kraken_key,
secret_loader.load_or_fail('KRAKEN_SECRET'),
),
)
del kraken_key
if kucoin_key := secret_loader.load('KUCOIN_KEY'):
depo_fetchers.append(
fin_depo.defi_kucoin.KucoinDepoFetcher(
kucoin_key,
secret_loader.load_or_fail('KUCOIN_SECRET'),
secret_loader.load_or_fail('KUCOIN_PASS'),
),
)
del kucoin_key
return fin_depo.static.AggregateDepoFetcher('Aggregated', depo_fetchers)
def _format_depo_tree_internal(
depo: fin_depo.data.Depo,
fmt: list[str],
indent: str,
more_whitespace: bool = False,
):
"""Internal string formatting for `format_depo_tree`."""
fmt.append(depo.name)
fmt.append('\n')
if isinstance(depo, fin_depo.data.DepoGroup):
for idx, nested_depo in enumerate(depo.nested):
is_last = idx == len(depo.nested) - 1
fmt.append(indent)
indent_new = indent + (' ' if is_last else '') + ' '
line = '└─ ' if is_last else '├─ '
fmt.append(line)
_format_depo_tree_internal(nested_depo, fmt, indent_new)
else:
depo_assets = list(depo.assets())
for idx, asset in enumerate(depo_assets):
is_last = idx == len(depo_assets) - 1
amount = depo.get_amount_of_asset(asset)
line = '└─ ' if is_last else '├─ '
derp = indent + line + asset.raw_short_name()
amount_str = f'{amount: 25.10f}'.rstrip('0').removesuffix('.')
fmt.append(f'{derp:20s}{amount_str}\n')
if more_whitespace:
fmt.append(f'{indent}\n')
def format_depo_tree(
depo: fin_depo.data.Depo,
more_whitespace: bool = False,
) -> str:
"""Formats the given `Depo` to a tree structure."""
fmt: list[str] = []
_format_depo_tree_internal(depo, fmt, ' ', more_whitespace=more_whitespace)
return ''.join(fmt)
def main():
"""Main function."""
aggregated_depos = setup_aggregate_depos()
depo = aggregated_depos.get_depo()
sys.stdout.write(format_depo_tree(depo))
if __name__ == '__main__':
main()

View File

@ -1,3 +1,5 @@
"""Datastructures for `fin-depo`."""
import abc
import dataclasses
import datetime
@ -12,7 +14,7 @@ from fin_defs import Asset, AssetAmount
@enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True)
class Depo(abc.ABC):
"""A depository tracking some amount of assets.
"""A depository tracking several assets.
Depo can either be DepoSingle, which is the base layer of the depository
structure, or nested in DepoGroup, which allows for a complex hierarcy of