This should conclude the arbiter code for those that have their own currency and wish for it to maintain a certain value, or to increase its value. It would be the same as investing in any other dex to bring your token value and have it become liquid so it becomes utilized.
- Arbiter Bot - Part 1 - arbiter_bot.py
- Arbiter Bot - Part 2 - arbiter_cancel.py
- Arbiter Bot - Part 3 - fetch_market.py
import requests
from beem import Hive
from datetime import datetime
API_URL = "https://api.hive-engine.com/rpc/contracts"
def format_decimal(value, places=8):
return f"{float(value):.{places}f}"
def safe_float(value):
try:
return float(value)
except Exception:
return 0.0
def _parse_timestamp(value):
if value is None:
return None
if isinstance(value, (int, float)):
return float(value)
if isinstance(value, str):
raw = value.strip()
if not raw:
return None
try:
return float(raw)
except Exception:
pass
try:
if raw.endswith("Z"):
raw = raw[:-1] + "+00:00"
return datetime.fromisoformat(raw).timestamp()
except Exception:
return None
return None
def _order_age_key(order):
timestamp_candidates = [
order.get("timestamp"),
order.get("time"),
order.get("createdAt"),
order.get("created"),
]
for candidate in timestamp_candidates:
parsed = _parse_timestamp(candidate)
if parsed is not None:
return (0, parsed)
block_candidates = [
order.get("blockNumber"),
order.get("block"),
]
for candidate in block_candidates:
try:
return (1, int(candidate))
except Exception:
continue
fallback_id = str(order.get("_id") or order.get("txId") or "")
return (2, fallback_id)
def place_order(account_name, token, price, quantity, order_type="buy", active_key=None, nodes=None):
hive = Hive(keys=[active_key], node=nodes or ["https://api.hive.blog"])
contract_payload = {
"contractName": "market",
"contractAction": "buy" if order_type == "buy" else "sell",
"contractPayload": {
"symbol": token,
"quantity": format_decimal(quantity),
"price": format_decimal(price),
},
}
try:
hive.custom_json(
id="ssc-mainnet-hive",
json_data=contract_payload,
required_auths=[account_name],
required_posting_auths=[],
)
print(f"[DEBUG] Placed {order_type.upper()} order: {quantity} {token} @ {price}")
return True
except Exception as exc:
print(f"[ERROR] Failed to place {order_type} order: {exc}")
return False
def get_open_orders(account_name, token=None):
all_orders = []
for table in ["buyBook", "sellBook"]:
offset = 0
page_size = 1000
while True:
query = {"account": account_name}
if token:
query["symbol"] = token
payload = {
"jsonrpc": "2.0",
"method": "find",
"params": {
"contract": "market",
"table": table,
"query": query,
"limit": page_size,
"offset": offset,
},
"id": 1,
}
try:
response = requests.post(API_URL, json=payload, timeout=10)
response.raise_for_status()
orders = response.json().get("result", [])
for order in orders:
order["type"] = "buy" if table == "buyBook" else "sell"
all_orders.extend(orders)
if len(orders) < page_size:
break
offset += page_size
except Exception as exc:
print(f"[ERROR] Failed to fetch {table} orders: {exc}")
break
return all_orders
def get_balance(account_name, token):
payload = {
"jsonrpc": "2.0",
"method": "findOne",
"params": {
"contract": "tokens",
"table": "balances",
"query": {"account": account_name, "symbol": token},
},
"id": 1,
}
try:
response = requests.post(API_URL, json=payload, timeout=10)
response.raise_for_status()
result = response.json().get("result")
return float(result["balance"]) if result and "balance" in result else 0.0
except Exception as exc:
print(f"[ERROR] Failed to fetch balance for {token}: {exc}")
return 0.0
def cancel_order(account_name, order_type, order_id, active_key=None, nodes=None):
hive = Hive(keys=[active_key], node=nodes or ["https://api.hive.blog"])
payload = [{
"contractName": "market",
"contractAction": "cancel",
"contractPayload": {
"type": order_type,
"id": str(order_id),
},
}]
try:
hive.custom_json(
id="ssc-mainnet-hive",
json_data=payload,
required_auths=[account_name],
required_posting_auths=[],
)
print(f"[DEBUG] Canceled {order_type} order {order_id}")
return True
except Exception as exc:
print(f"[ERROR] Failed to cancel order {order_id}: {exc}")
return False
def cancel_oldest_order(account_name, active_key=None, nodes=None, token=None):
orders = get_open_orders(account_name, token=token)
if not orders:
print("[DEBUG] No open orders to cancel")
return False
oldest = min(orders, key=_order_age_key)
order_id = oldest.get("txId")
order_type = oldest.get("type")
if not order_id or order_type not in {"buy", "sell"}:
print("[ERROR] Could not determine order id/type for cancellation")
return False
return cancel_order(account_name, order_type, order_id, active_key=active_key, nodes=nodes)
🤖 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.