222 lines
7.1 KiB
Python
222 lines
7.1 KiB
Python
"""# Automatic Crypto Seller.
|
|
|
|
This is a wrapper about [`fin_depo`] for automatically individual financial
|
|
assets a one way, by using repeated trades over a period.
|
|
|
|
The primary motivation is for trading low-volume crypto assets slowly without
|
|
affecting the price too much.
|
|
|
|
Supported sites:
|
|
|
|
- [KuCoin](https://www.kucoin.com/): Online crypto-currency exchange.
|
|
|
|
## Installation
|
|
|
|
Install dependencies:
|
|
|
|
```shell
|
|
pip install -r requirements.txt
|
|
```
|
|
|
|
## Usage
|
|
|
|
Run using from the top directory:
|
|
|
|
```shell
|
|
python -m crypto_seller
|
|
```
|
|
|
|
The script will now automatically sell assets over time. The interval between
|
|
sell-offs are randomized to avoid front-running, and the amounts are also
|
|
randomized to avoid too consistent behaviour.
|
|
|
|
The log will both be displayed in the shell, and be placed in a log file
|
|
`output/log.txt`. Every sell-off will be logged to the `output/trades.csv`
|
|
file, with as much information as possible. Keep both of these for tax
|
|
purposes, if relevant.
|
|
|
|
## Auditing information
|
|
|
|
The most relevant libraries for auditing are:
|
|
|
|
- [`fin_depo`](https://gitfub.space/Jmaa/fin-depo): Library for programmatic
|
|
fetching of depository assets and in some cases allows for order placement.
|
|
This is the library that reads balances and places market orders.
|
|
- [`fin_defs`](https://gitfub.space/Jmaa/fin-defs): Definitions of financial
|
|
assets and instruments. Used by `fin_depo`.
|
|
- [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) is used by
|
|
`fin_depo` for providing KuCoin support.
|
|
|
|
## Taxation
|
|
|
|
Some parts of the world suffers from weird and draconian taxation rules on
|
|
cryptocurrencies, so it helps to keep track of the relevant trading
|
|
information.
|
|
|
|
As mentioned above, keep (and backup) both the log file (`output/log.txt`) and
|
|
the trades file (`output/trades.csv`).
|
|
|
|
To help with tax reporting, it might be useful to sign up to tax oriented
|
|
websites. For example, [CryptoSkat](https://cryptoskat.dk/) seems to be the
|
|
most mature on the danish market, and does support KuCoin.
|
|
|
|
## TODO
|
|
|
|
- [ ] Parse configuration from json.
|
|
- [ ] Ensure that a failure during selling results in a safe winding down of the system.
|
|
* Catch runtime errors when selling
|
|
* Show errors to log.
|
|
* Stop loop and exit with results, and error indicator.
|
|
- [ ] Document configuration
|
|
- [ ] Document auditing
|
|
- [X] Log all trades to CSV file.
|
|
- [X] Collect information during the run and output after run
|
|
|
|
"""
|
|
|
|
__all__ = ['__version__', 'run_auto_sell']
|
|
|
|
import dataclasses
|
|
import datetime
|
|
import logging
|
|
import random
|
|
from collections.abc import Callable
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import fin_defs
|
|
import fin_depo
|
|
|
|
from ._version import __version__
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
################################################################################
|
|
# Main code #
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class AutoSellConfig:
|
|
"""Configuration for `run_auto_sell`.
|
|
|
|
Fields:
|
|
|
|
- `interval_range`: The range to determine sell-off intervals from. After
|
|
each sell-off it will sleep an amount of time uniformly sampled from this
|
|
range.
|
|
- `input_asset`: Asset to sell.
|
|
- `input_amount_range`: Range to uniformly sample sell-off sizes from.
|
|
- `output_asset`: Asset to "buy".
|
|
- `seller`: Depository seller to use.
|
|
- `sleep`: Sleep method to use.
|
|
- `exit_when_empty`: Whether the system should stop once there is nothing
|
|
more to sell. Recommended.
|
|
"""
|
|
|
|
interval_range: tuple[datetime.timedelta, datetime.timedelta]
|
|
input_asset: fin_defs.Asset
|
|
input_amount_range: tuple[Decimal, Decimal]
|
|
output_asset: fin_defs.Asset
|
|
seller: fin_depo.defi_kucoin.KucoinDepoFetcher
|
|
sleep: Callable[[float], None]
|
|
log_order_to_csv: Callable[[fin_depo.data.TradeOrderDetails], None]
|
|
exit_when_empty: bool = True
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True)
|
|
class AutoSellRunResults:
|
|
"""Result information after `run_auto_sell`.
|
|
|
|
Fields:
|
|
- `order_details`: The list of order details.
|
|
- `total_duration`: Total duration of the run.
|
|
- `total_sleep_duration`: Amount of time spent waiting between sell-offs.
|
|
"""
|
|
|
|
order_details: list[fin_depo.data.TradeOrderDetails]
|
|
total_duration: datetime.timedelta
|
|
total_sleep_duration: datetime.timedelta
|
|
|
|
def total_input_amount(self) -> Decimal:
|
|
"""Total amount of assets sold."""
|
|
return sum(o.input_amount for o in self.order_details)
|
|
|
|
def total_output_amount(self) -> Decimal:
|
|
"""Total amount of assets "bought"."""
|
|
return sum(o.output_amount for o in self.order_details)
|
|
|
|
|
|
def sample_from_range(rng: random.Random, range: tuple[Any, Any]) -> Any:
|
|
"""Samples uniformly from the given range."""
|
|
multiplier: float | Decimal = rng.random()
|
|
if isinstance(range[0], Decimal):
|
|
multiplier = Decimal(multiplier)
|
|
return range[0] + (range[1] - range[0]) * multiplier
|
|
|
|
|
|
def run_auto_sell(config: AutoSellConfig) -> AutoSellRunResults:
|
|
"""Executes the sell-off.
|
|
|
|
Sell-offs are performed in rounds of sizes and with intervals randomly
|
|
based on the given configuration.
|
|
"""
|
|
rng = random.SystemRandom()
|
|
|
|
time_start = datetime.datetime.now(tz=datetime.UTC)
|
|
all_executed_orders: list[fin_depo.data.TradeOrderDetails] = []
|
|
total_sleep_duration = datetime.timedelta(seconds=0)
|
|
|
|
while True:
|
|
# Check that account has tokens.
|
|
input_amount_available = config.seller.get_depo().get_amount_of_asset(
|
|
config.input_asset,
|
|
)
|
|
logger.info('Currently own %s %s', input_amount_available, config.input_asset)
|
|
|
|
if input_amount_available > 0:
|
|
amount_to_sell = sample_from_range(rng, config.input_amount_range)
|
|
amount_to_sell = min(input_amount_available, amount_to_sell)
|
|
logger.info('Attempting to sell %s %s', amount_to_sell, config.input_asset)
|
|
|
|
order_details = config.seller.place_market_order(
|
|
config.input_asset,
|
|
amount_to_sell,
|
|
config.output_asset,
|
|
)
|
|
|
|
config.log_order_to_csv(order_details)
|
|
all_executed_orders.append(order_details)
|
|
|
|
del amount_to_sell
|
|
elif config.exit_when_empty:
|
|
break
|
|
|
|
# Time out
|
|
time_to_sleep = sample_from_range(rng, config.interval_range)
|
|
time_to_sleep_secs = time_to_sleep.total_seconds()
|
|
logger.info('Sleeping %s (%d seconds)', time_to_sleep, time_to_sleep_secs)
|
|
config.sleep(time_to_sleep_secs)
|
|
total_sleep_duration += time_to_sleep
|
|
|
|
del input_amount_available
|
|
|
|
time_end = datetime.datetime.now(tz=datetime.UTC)
|
|
|
|
return AutoSellRunResults(
|
|
all_executed_orders,
|
|
total_duration=time_end - time_start,
|
|
total_sleep_duration=total_sleep_duration,
|
|
)
|
|
|
|
|
|
def log_results(results: AutoSellRunResults):
|
|
logger.info('Stats:')
|
|
logger.info('- Num orders: %s', len(results.order_details))
|
|
logger.info('- Total sold: %s', results.total_input_amount())
|
|
logger.info('- Total gained: %s', results.total_output_amount())
|
|
logger.info(
|
|
'- Total duration: %s (%s spent sleeping)',
|
|
results.total_duration,
|
|
results.total_sleep_duration,
|
|
)
|