Implement Coinbase Advanced API to replace Historic Crypto Plugin

[macanudo527]

The current Historic Cypto Plugin Silently exits due to an incorrect API call. The Historic Crypto module appears to be abandoned, so we need to implement a better solution. The Coinbase Advanced API should offer a better long term solution.

[ndopencode]

I noticed this issue recently. Is there a work around while Historic Crypto is no longer working?

[macanudo527]

Not really. It should be pretty simple to implement the Coinbase Advanced API if you want to submit something though.

[orientalperil]

I made something for my use that does the same thing as Historic Crypto but uses Coinbase Advanced API. I'm using AutoConfig from python-decouple to read the api key from an .env file.

from datetime import datetime
from datetime import timedelta
from datetime import timezone
from typing import Optional

from coinbase.rest import RESTClient
from dali.abstract_pair_converter_plugin import AbstractPairConverterPlugin
from dali.historical_bar import HistoricalBar
from dali.transaction_manifest import TransactionManifest
from rp2.rp2_decimal import RP2Decimal

from dali_dali.settings import config


class PairConverterPlugin(AbstractPairConverterPlugin):
    def __init__(self, historical_price_type: str) -> None:
        super().__init__(historical_price_type)
        self.client = RESTClient(api_key=config('COINBASE_API_KEY'), api_secret=config('COINBASE_PRIVATE_KEY'))

    def name(self) -> str:
        return "dali_dali_coinbase"

    def cache_key(self) -> str:
        return self.name()

    def optimize(self, transaction_manifest: TransactionManifest) -> None:
        pass

    def get_historic_bar_from_native_source(self, timestamp: datetime, from_asset: str, to_asset: str, exchange: str) -> Optional[HistoricalBar]:
        time_granularity = 'ONE_MINUTE'
        utc_timestamp = timestamp.astimezone(timezone.utc)
        start = utc_timestamp.replace(second=0)
        end = start

        candle = self.client.get_candles(f'{from_asset}-{to_asset}', int(start.timestamp()), int(end.timestamp()), time_granularity).to_dict()['candles'][0]
        result = HistoricalBar(
            duration=timedelta(seconds=60),
            timestamp=start,
            open=RP2Decimal(candle['open']),
            high=RP2Decimal(candle['high']),
            low=RP2Decimal(candle['low']),
            close=RP2Decimal(candle['close']),
            volume=RP2Decimal(candle['volume']),
        )

        return result

[macanudo527]

@orientalperil I refactored your code and created a plugin with it. If you have the time can you test it out? You can find it in #271

Thanks for contributing the code sample!