normanlmfung

NODE_URL = "https://fullnode.mainnet.aptoslabs.com/v1"
FAUCET_URL = "https://tap.devnet.prod.gcp.aptosdev.com"

wallet_address = None
private_key = None

rest_client = HoustonClient(NODE_URL)
pk = PrivateKey.from_hex(private_key)
addr = AccountAddress.from_hex(wallet_address)
acc = Account(addr, pk)

normanlmfung

import time
from typing import Optional

from aptos_sdk.account import Account
from aptos_sdk.account_address import AccountAddress
from aptos_sdk.bcs import Serializer
from aptos_sdk.client import FaucetClient, RestClient
from aptos_sdk.transactions import (
    EntryFunction,
    TransactionArgument,

normanlmfung

from aptos_sdk.account import Account, AccountAddress
from aptos_sdk.client import RestClient, FaucetClient
from aptos_sdk.ed25519 import PrivateKey

NODE_URL = "https://fullnode.devnet.aptoslabs.com/v1"
FAUCET_URL = "https://tap.devnet.prod.gcp.aptosdev.com"

wallet_address = "xxx"
private_key = "xxx"

normanlmfung

import time
from datetime import datetime
from web3 import Web3

RATELIMIT_MESSAGES = [
    "Too Many Requests"
]

# https://www.cronosresearch.com/resources/rpc-list
rpc_providers = [

normanlmfung

"""
This script gets the buy/sell prices from the vvs finance (https://vvs.finance/swap) smart contracts
for a given block number and path.

"""

from datetime import datetime

from dataclasses import dataclass
from typing import (

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

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 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

'''
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

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