Arbiter Bot - Part 3 - fetch_market.py

• Arbiter Bot - Part 1 - arbiter_bot.py
• Arbiter Bot - Part 2 - arbiter_cancel.py

Just hang in there with me.

I’ve been putting time into working on @enginewitty and the P.I.M.P. machine, and I know I have not been saying much while doing it.

Sometimes, after I get done writing the code, I just do not have much left in me to type out a full explanation.

This part is fetch_market.py.

Like the other bots, this file is built to watch the Hive Engine market and pull the numbers the bot needs before making decisions. It checks the buy book and sell book, finds the current highest bid and lowest ask, and gives the bot a quick snapshot of the market so it can react to live conditions. It also includes helpers for pulling open orders from an account when needed.

import requests

API_URL = "https://api.hive-engine.com/rpc/contracts"


def _query_book(table, token, limit=1000, descending=False):
    index = "priceDec" if table == "buyBook" else "price"
    payload = {
        "jsonrpc": "2.0",
        "method": "find",
        "params": {
            "contract": "market",
            "table": table,
            "query": {"symbol": token},
            "limit": limit,
            "indexes": [{"index": index, "descending": descending}],
        },
        "id": 1,
    }
    response = requests.post(API_URL, json=payload, timeout=10)
    response.raise_for_status()
    return response.json().get("result", [])


def get_orderbook_top(token="ARBITER"):
    try:
        buy_result = _query_book("buyBook", token, descending=True)
        sell_result = _query_book("sellBook", token, descending=False)

        highest_bid = float(buy_result[0]["price"]) if buy_result else 0.0
        valid_asks = [float(order["price"]) for order in sell_result if float(order.get("price", 0)) > 0]
        lowest_ask = min(valid_asks) if valid_asks else 0.0

        return {"highestBid": highest_bid, "lowestAsk": lowest_ask}
    except Exception as exc:
        print(f"[ERROR] Failed to fetch top orderbook for {token}: {exc}")
        return None


def get_orderbook_depth(token="ARBITER", depth=50):
    try:
        buy_result = _query_book("buyBook", token, limit=depth, descending=True)
        sell_result = _query_book("sellBook", token, limit=depth, descending=False)

        buy_prices = [float(o["price"]) for o in buy_result if float(o.get("price", 0)) > 0]
        sell_prices = [float(o["price"]) for o in sell_result if float(o.get("price", 0)) > 0]
        return buy_prices, sell_prices
    except Exception as exc:
        print(f"[ERROR] Failed to fetch orderbook depth for {token}: {exc}")
        return [], []

🤖 PeakeBot — Autonomous Trading System (RC-AWARE)

Independent multi-token trading bot featuring:
RC-aware execution, adaptive delay logic, and self-regulating trade cycles.

📊 Trading bot details:
👉 https://geocities.ws/p/e/peakecoin/trading-bot/peakebot_v0_01.html
💻 Open-source repositories:
👉 https://github.com/paulmoon410

🙏 Acknowledgements

Thanks to and please follow:
@enginewitty @ecoinstant @neoxian @txracer @thecrazygm @holdonia @aggroed

For their continued support, guidance, and help in expanding the PeakeCoin ecosystem.