Jmaa/personal-data
1
0
Fork 0
Code Actions Packages

Ruff
Some checks failed
Test Python / Test (push) Failing after 25s
Details

Browse Source
This commit is contained in:
Jon Michael Aanes 2024-06-02 18:01:18 +02:00
parent ff6e5110a4
commit 9d992abb44
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
4 changed files with 17 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
personal_data/fetchers/defi_kucoin.py
View File

@ -1,13 +1,12 @@
import abc
import dataclasses import dataclasses
import datetime import datetime
import logging import logging
from collections.abc import Iterator, Mapping from collections.abc import Iterator, Mapping
from decimal import Decimal
import abc
import fin_depo
import kucoin.client import kucoin.client
from frozendict import frozendict from frozendict import frozendict
import fin_depo
from personal_data.data import DeduplicateMode, Scraper from personal_data.data import DeduplicateMode, Scraper
@ -23,9 +22,9 @@ client = kucoin.client.Client(
secrets.KUCOIN_PASS, secrets.KUCOIN_PASS,
) )
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class FinanceDepoScraper(Scraper): class FinanceDepoScraper(Scraper):
@abc.abstractmethod @abc.abstractmethod
def get_depo_fetcher(self): def get_depo_fetcher(self):
pass pass
@ -38,7 +37,7 @@ class FinanceDepoScraper(Scraper):
} }
for asset in depo.assets(): for asset in depo.assets():
key = f"balance.{asset}" key = f'balance.{asset}'
data_point[key] = depo.get_amount_of_asset(asset) data_point[key] = depo.get_amount_of_asset(asset)
print(data_point) print(data_point)
@ -54,10 +53,11 @@ class KrakenAccountBalances(FinanceDepoScraper):
def get_depo_fetcher(self): def get_depo_fetcher(self):
return fin_depo.defi_kraken.KrakenDepoFetcher( return fin_depo.defi_kraken.KrakenDepoFetcher(
kraken_key = secrets.KRAKEN_KEY, kraken_key=secrets.KRAKEN_KEY,
kraken_secret = secrets.KRAKEN_SECRET, kraken_secret=secrets.KRAKEN_SECRET,
) )
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class KucoinAccountBalances(FinanceDepoScraper): class KucoinAccountBalances(FinanceDepoScraper):
dataset_name = 'defi_kucoin_balance' dataset_name = 'defi_kucoin_balance'
@ -66,12 +66,13 @@ class KucoinAccountBalances(FinanceDepoScraper):
def get_depo_fetcher(self): def get_depo_fetcher(self):
return fin_depo.defi_kucoin.KucoinDepoFetcher( return fin_depo.defi_kucoin.KucoinDepoFetcher(
kucoin_key = secrets.KUCOIN_KEY, kucoin_key=secrets.KUCOIN_KEY,
kucoin_secret = secrets.KUCOIN_SECRET, kucoin_secret=secrets.KUCOIN_SECRET,
kucoin_pass = secrets.KUCOIN_PASS, kucoin_pass=secrets.KUCOIN_PASS,
) )
def addresses_to_data_points(addresses: list[dict[str,str]]) -> frozendict:
def addresses_to_data_points(addresses: list[dict[str, str]]) -> frozendict:
data_point = {} data_point = {}
data_point['account.update_time'] = datetime.datetime.now(tz=datetime.UTC) data_point['account.update_time'] = datetime.datetime.now(tz=datetime.UTC)
data_point['account.num_deposit_addresses'] = len(addresses) data_point['account.num_deposit_addresses'] = len(addresses)

2
personal_data/main.py
View File

@ -1,7 +1,7 @@
import csv import csv
import datetime import datetime
import inspect
import decimal import decimal
import inspect
import io import io
import logging import logging
from collections.abc import Iterable, Mapping, Sequence from collections.abc import Iterable, Mapping, Sequence

3
setup.py
View File

@ -15,6 +15,7 @@ PACKAGE_NAME = 'personal_data'
with open('README.md') as f: with open('README.md') as f:
readme = f.read() readme = f.read()
def parse_version_file(text: str) -> str: def parse_version_file(text: str) -> str:
match = re.match(r'^__version__\s*=\s*(["\'])([\d\.]+)\1$', text) match = re.match(r'^__version__\s*=\s*(["\'])([\d\.]+)\1$', text)
if match is None: if match is None:
@ -22,9 +23,11 @@ def parse_version_file(text: str) -> str:
raise Exception(msg) raise Exception(msg)
return match.group(2) return match.group(2)
with open(PACKAGE_NAME + '/_version.py') as f: with open(PACKAGE_NAME + '/_version.py') as f:
version = parse_version_file(f.read()) version = parse_version_file(f.read())
def parse_requirements(text: str) -> list[str]: def parse_requirements(text: str) -> list[str]:
return text.strip().split('\n') return text.strip().split('\n')