normanlmfung

from decimal import Decimal
from typing import Dict

from dotmap import DotMap
import ccxt

class binanceperps(ccxt.binance):
    def describe(self) -> Dict:
        description = super().describe()
        description["options"]["defaultType"] = "future"

normanlmfung

import json
import asyncio
from web3 import Web3

'''
SINGLE/USDC LP: https://vvs.finance/info/farm/0x0fbab8a90cac61b481530aad3a64fe17b322c25d
This will lead to CRONOS Explorer (Where you'd find ABI and contract source code): https://cronos.org/explorer/address/0x0fBAB8A90CAC61b481530AAd3a64fE17B322C25d/contracts
Cronos integration doc (Where you'd find the node_url): https://cronos.org/docs/resources/chain-integration.html 
'''
from web3 import Web3

normanlmfung

from datetime import datetime
import json
from web3 import Web3

my_wallet_address = "xxx"
wallet_address = Web3.toChecksumAddress(my_wallet_address)
private_key = "xxx"

cronos_mainnet_rpc = "https://evm.cronos.org"
w3 = Web3(Web3.HTTPProvider(cronos_mainnet_rpc))

normanlmfung

from datetime import datetime, timedelta
import pytz
import arrow
from typing import Dict
import pandas as pd
from ccxt import deribit, binance

spot_exchange = binance()
btc_usdt_ob = spot_exchange.fetch_order_book('BTC/USDT')
best_ask = min([x[0] for x in btc_usdt_ob['asks']])

normanlmfung

# Sample from @Kroitor 20220426
import ccxt

def table(values):
    first = values[0]
    keys = list(first.keys()) if isinstance(first, dict) else range(0, len(first))
    widths = [max([len(str(v[k])) for v in values]) for k in keys]
    string = ' | '.join(['{:<' + str(w) + '}' for w in widths])
    return "\n".join([string.format(*[str(v[k]) for k in keys]) for v in values])

normanlmfung

from typing import Dict
from web3 import Web3
from web3.exceptions import TransactionNotFound

ONE_BILLION : int = 1000000000
EIP20_ABI = '[{"constant":true,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"_spender","type":"address"},{"name":"_value","type":"uint256"}],"name":"approve","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"totalSupply","outputs":[{"name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"name":"_from","type":"address"},{"name":"_to","type":"address"},{"name":"_value","type":"uint256"}],"name":"transferFrom","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"payable":false,"stateMuta

normanlmfung

'''
This is taken from https://web3py.readthedocs.io/en/stable/examples.html#advanced-example-fetching-all-token-transfer-events

The following changes have been made so we now fetches all Transfer under SINGLE token on CRONOS chain, after an arbitrary "cutoff":
a) TARGET_TOKEN_ADDRESS (Replacing RCC_ADDRESS in original script) now referencing SINGLE token on CRONOS chain
b) api_url = "https://evm.cronos.org"
c) We're only processing event_type = Transfer. Look at "scan_chunk", it'd only call "_fetch_events_for_all_contracts" for event_type==Transfer
    scanner = EventScanner(
            web3=web3,
            contract=ERC20,

normanlmfung

import logging
import datetime
from requests import ReadTimeout
from web3.exceptions import BlockNotFound

logger = logging.getLogger(__name__)

def search_block_with_cutoff_ts(
                                    end_block : int, 
                                    cutoff_ts : int, 

normanlmfung

import unittest
import datetime
from typing import List

from src.gizmo.blockchain_gizmo import search_block_with_cutoff_ts

class BlockChainGizmoTests(unittest.TestCase):
    def test_search_block_with_cutoff_ts(self): 
        class DummyBlock:
            def __init__(self, _number : int, _timestamp : int) -> None:

normanlmfung

import time
import enum
import asyncio
import nest_asyncio
from core.base_types import DenormalizedTicker, NormalizedTicker
from typing import List, Dict
from ccxt.base.exchange import Exchange
from ccxt.ftx import ftx
from ccxt.binance import binance
from ccxt.huobi import huobi