I've received constant feedback, people aren't interested in the games.
They want to make money here on Hive.
They want their tokens to make money here on Hive.
Well, here's the beginning if you feed the bot, it will raise the prices.
There will be several other files. I'll try to keep them all linked together, so if you read the next one you will at least find this one.
import os
import json
import subprocess
import time
from datetime import datetime
from beem import Hive
from fetch_market import get_orderbook_top
from place_order import get_balance, place_order, get_open_orders, safe_float
DEFAULT_CONFIG = {
"account_name": "YOUR_HIVE_ACCOUNT",
"active_key": "YOUR_ACTIVE_KEY",
"token": "ARBITER",
"quote_token": "SWAP.HIVE",
"target_price": 0.10,
"spread": 0.004,
"sell_allocation_pct": 0.10,
"buy_allocation_pct": 0.50,
"min_quote_for_buy": 0.01,
"min_profit_spread_pct": 0.004,
"edge_buffer_pct": 0.001,
"duplicate_price_tolerance_pct": 0.001,
"min_rc_percent": 10,
"low_rc_cycle_seconds": 1800,
"initial_token_reserve": 1.0,
"reserve_growth_per_cycle": 0.00000001,
"cycle_seconds": 900,
"cancel_every_n_cycles": 3,
"hive_nodes": ["https://api.hive.blog", "https://anyx.io"],
}
def read_bot_info(path="bot_info.txt"):
try:
with open(path, "r", encoding="utf-8") as file:
lines = [line.strip() for line in file if line.strip()]
if len(lines) >= 2:
return lines[0], lines[1]
except Exception:
pass
return None, None
def run_cancel_once(script_path):
try:
subprocess.run(["python", script_path, "--once"], check=False)
except Exception as exc:
print(f"[WARN] Cancel helper failed: {exc}")
def safe_price(value):
try:
v = float(value)
return v if v > 0 else 0.0
except Exception:
return 0.0
def clamp_01(value, default_value):
try:
v = float(value)
except Exception:
return default_value
return min(max(v, 0.0), 1.0)
def safe_int(value, default_value, minimum=1):
try:
iv = int(value)
except Exception:
iv = default_value
return max(iv, minimum)
def normalize_config(config):
normalized = dict(config)
normalized["target_price"] = safe_price(config.get("target_price"))
normalized["spread"] = safe_price(config.get("spread"))
normalized["min_quote_for_buy"] = safe_price(config.get("min_quote_for_buy"))
normalized["min_profit_spread_pct"] = safe_price(config.get("min_profit_spread_pct"))
normalized["edge_buffer_pct"] = safe_price(config.get("edge_buffer_pct"))
normalized["duplicate_price_tolerance_pct"] = safe_price(config.get("duplicate_price_tolerance_pct"))
normalized["min_rc_percent"] = safe_int(config.get("min_rc_percent"), 10, minimum=1)
normalized["initial_token_reserve"] = safe_price(config.get("initial_token_reserve"))
normalized["reserve_growth_per_cycle"] = safe_price(config.get("reserve_growth_per_cycle"))
normalized["sell_allocation_pct"] = clamp_01(
config.get("sell_allocation_pct"),
DEFAULT_CONFIG["sell_allocation_pct"],
)
normalized["buy_allocation_pct"] = clamp_01(
config.get("buy_allocation_pct"),
DEFAULT_CONFIG["buy_allocation_pct"],
)
normalized["cycle_seconds"] = safe_int(
config.get("cycle_seconds"),
DEFAULT_CONFIG["cycle_seconds"],
minimum=5,
)
normalized["cancel_every_n_cycles"] = safe_int(
config.get("cancel_every_n_cycles"),
DEFAULT_CONFIG["cancel_every_n_cycles"],
minimum=1,
)
normalized["low_rc_cycle_seconds"] = safe_int(
config.get("low_rc_cycle_seconds"),
DEFAULT_CONFIG["low_rc_cycle_seconds"],
minimum=5,
)
nodes = config.get("hive_nodes")
if not isinstance(nodes, list) or not nodes:
nodes = DEFAULT_CONFIG["hive_nodes"]
normalized["hive_nodes"] = nodes
return normalized
def setup_config_interactive(config_path):
print("[SETUP] First-run setup for ARBITER bot")
account_name = input("Hive account name: ").strip()
active_key = input("Active key: ").strip()
token = input("Token to market-make (default ARBITER): ").strip() or "ARBITER"
target_price = 0.0
while target_price <= 0:
raw = input("Target price to maintain in SWAP.HIVE (e.g. 0.10): ").strip()
target_price = safe_price(raw)
if target_price <= 0:
print("[SETUP] Target price must be greater than 0.")
config = dict(DEFAULT_CONFIG)
config.update(
{
"account_name": account_name,
"active_key": active_key,
"token": token,
"target_price": target_price,
}
)
with open(config_path, "w", encoding="utf-8") as file:
json.dump(config, file, indent=2)
print(f"[SETUP] Saved {config_path}")
return config
def load_config(root):
config_path = os.path.join(root, "bot_config.json")
legacy_bot_info_path = os.path.join(root, "bot_info.txt")
if os.path.exists(config_path):
try:
with open(config_path, "r", encoding="utf-8") as file:
user_config = json.load(file)
config = dict(DEFAULT_CONFIG)
config.update(user_config if isinstance(user_config, dict) else {})
return config
except Exception as exc:
print(f"[ERROR] Failed to read bot_config.json: {exc}")
raise SystemExit(1)
account, active_key = read_bot_info(legacy_bot_info_path)
if account and active_key:
config = dict(DEFAULT_CONFIG)
config.update({"account_name": account, "active_key": active_key})
with open(config_path, "w", encoding="utf-8") as file:
json.dump(config, file, indent=2)
print("[MIGRATE] Converted bot_info.txt to bot_config.json")
return config
return setup_config_interactive(config_path)
def build_prices(target_price, spread, highest_bid, lowest_ask):
half_spread = max(spread / 2.0, 0.00000001)
buy_price = max(target_price - half_spread, 0.00000001)
sell_price = max(target_price + half_spread, buy_price + 0.00000001)
# Keep the orders near current book edges while staying around the target.
if highest_bid > 0:
buy_price = min(buy_price, highest_bid + half_spread)
if lowest_ask > 0:
sell_price = max(sell_price, lowest_ask - half_spread)
if sell_price <= buy_price:
sell_price = buy_price + 0.00000001
return buy_price, sell_price
def build_guarded_prices(
target_price,
spread,
highest_bid,
lowest_ask,
edge_buffer_pct,
min_profit_spread_pct,
):
if highest_bid <= 0 or lowest_ask <= 0:
return None, None
live_spread = max(lowest_ask - highest_bid, 0.0)
min_target_spread = max(spread, min_profit_spread_pct, 0.00000002)
# If the book is inverted or too tight, widen around mid to keep margin.
if live_spread < min_target_spread:
mid = (highest_bid + lowest_ask) / 2.0
buy_price = max(mid - (min_target_spread / 2.0), 0.00000001)
sell_price = max(mid + (min_target_spread / 2.0), buy_price + 0.00000001)
else:
buy_price, sell_price = build_prices(target_price, spread, highest_bid, lowest_ask)
# Keep a small guard from top-of-book edges to avoid crossing/touching too tightly.
buy_ceiling = max(highest_bid * (1.0 - edge_buffer_pct), 0.00000001)
sell_floor = max(lowest_ask * (1.0 + edge_buffer_pct), 0.00000001)
buy_price = min(buy_price, buy_ceiling)
sell_price = max(sell_price, sell_floor)
if sell_price <= buy_price:
return None, None
effective_spread_pct = (sell_price - buy_price) / buy_price if buy_price > 0 else 0.0
if effective_spread_pct < min_profit_spread_pct:
return None, None
return buy_price, sell_price
def has_similar_open_order(open_orders, order_type, price, tolerance_pct):
if price <= 0:
return False
for order in open_orders:
if order.get("type") != order_type:
continue
existing_price = safe_float(order.get("price"))
if existing_price <= 0:
continue
delta = abs(existing_price - price) / price
if delta <= tolerance_pct:
return True
return False
def get_locked_amounts(open_orders):
locked_quote = 0.0
locked_token = 0.0
for order in open_orders:
qty = safe_float(order.get("quantity"))
price = safe_float(order.get("price"))
if qty <= 0 or price <= 0:
continue
if order.get("type") == "buy":
locked_quote += qty * price
elif order.get("type") == "sell":
locked_token += qty
return locked_quote, locked_token
def get_rc_percent(account_name, active_key, nodes):
try:
hive = Hive(keys=[active_key], node=nodes)
acct = hive.get_account(account_name)
manabar = acct.get_rc_manabar()
current_pct = safe_float(manabar.get("current_pct"))
return max(current_pct, 0.0)
except Exception as exc:
print(f"[WARN] RC lookup failed: {exc}")
return None
def main():
root = os.path.dirname(__file__)
config = normalize_config(load_config(root))
account = (config.get("account_name") or "").strip()
active_key = (config.get("active_key") or "").strip()
token = (config.get("token") or "ARBITER").strip()
quote = (config.get("quote_token") or "SWAP.HIVE").strip()
target_price = config.get("target_price")
spread = config.get("spread")
sell_percentage = config.get("sell_allocation_pct")
buy_back_percentage = config.get("buy_allocation_pct")
min_quote_for_buy = config.get("min_quote_for_buy")
min_profit_spread_pct = config.get("min_profit_spread_pct")
edge_buffer_pct = config.get("edge_buffer_pct")
duplicate_price_tolerance_pct = config.get("duplicate_price_tolerance_pct")
min_rc_percent = config.get("min_rc_percent")
low_rc_cycle_seconds = config.get("low_rc_cycle_seconds")
token_reserve = config.get("initial_token_reserve")
reserve_growth_per_cycle = config.get("reserve_growth_per_cycle")
cycle_seconds = config.get("cycle_seconds")
cancel_every_n_cycles = config.get("cancel_every_n_cycles")
hive_nodes = config.get("hive_nodes")
if not account or not active_key or not token or target_price <= 0:
print("[ERROR] Invalid bot_config.json. account_name, active_key, token, and target_price are required.")
raise SystemExit(1)
cycle = 0
print(f"[BOOT] Starting market-maker for {token}/{quote} on account {account}")
print(f"[BOOT] Target price={target_price} {quote}, spread={spread}")
while True:
cycle += 1
token_reserve += reserve_growth_per_cycle
print(f"\n[{datetime.utcnow().isoformat()}Z] Cycle {cycle}")
rc_pct = get_rc_percent(account, active_key, hive_nodes)
if rc_pct is not None and rc_pct < min_rc_percent:
print(f"[RC] Low RC detected ({rc_pct:.2f}% < {min_rc_percent}%). Slowing cycle.")
time.sleep(low_rc_cycle_seconds)
continue
orderbook = get_orderbook_top(token)
if not orderbook:
print("[WARN] No orderbook data available; sleeping.")
time.sleep(cycle_seconds)
continue
highest_bid = safe_price(orderbook.get("highestBid"))
lowest_ask = safe_price(orderbook.get("lowestAsk"))
if highest_bid <= 0 or lowest_ask <= 0:
print(f"[WARN] Invalid spread: bid={highest_bid}, ask={lowest_ask}")
time.sleep(cycle_seconds)
continue
buy_price, sell_price = build_guarded_prices(
target_price,
spread,
highest_bid,
lowest_ask,
edge_buffer_pct,
min_profit_spread_pct,
)
if buy_price is None or sell_price is None:
print("[GUARD] Price guard rejected cycle (tight/inverted/unprofitable spread).")
time.sleep(cycle_seconds)
continue
open_orders = get_open_orders(account, token=token)
locked_quote, locked_token = get_locked_amounts(open_orders)
token_balance = get_balance(account, token)
hive_balance = get_balance(account, quote)
available_quote = max(hive_balance - locked_quote, 0.0)
available_token = max(token_balance - locked_token, 0.0)
available_to_sell = max(available_token - token_reserve, 0.0)
sell_qty = round(available_to_sell * sell_percentage, 8)
if sell_qty > 0 and not has_similar_open_order(
open_orders,
"sell",
sell_price,
duplicate_price_tolerance_pct,
):
print(f"[SELL] qty={sell_qty} {token} @ {sell_price}")
place_order(
account,
token,
sell_price,
sell_qty,
order_type="sell",
active_key=active_key,
nodes=hive_nodes,
)
else:
print(f"[SELL] Skipped (available_token={available_token}, reserve={token_reserve}, duplicate_check=true)")
hive_to_spend = available_quote * buy_back_percentage
buy_qty = round(hive_to_spend / buy_price, 8) if buy_price > 0 else 0.0
if (
hive_to_spend >= min_quote_for_buy
and buy_qty > 0
and not has_similar_open_order(
open_orders,
"buy",
buy_price,
duplicate_price_tolerance_pct,
)
):
print(f"[BUY] qty={buy_qty} {token} @ {buy_price} using {hive_to_spend:.8f} {quote}")
place_order(
account,
token,
buy_price,
buy_qty,
order_type="buy",
active_key=active_key,
nodes=hive_nodes,
)
else:
print(f"[BUY] Skipped (available_quote={available_quote}, min={min_quote_for_buy}, duplicate_check=true)")
if cancel_every_n_cycles > 0 and cycle % cancel_every_n_cycles == 0:
print("[CANCEL] Refreshing one old order")
run_cancel_once(os.path.join(root, "arbiter_cancel.py"))
print(f"[SLEEP] Waiting {cycle_seconds}s")
time.sleep(cycle_seconds)
if __name__ == "__main__":
main()
🤖 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 expanding the PeakeCoin ecosystem.