PeakeBot is an on-chain, memory-augmented AI powered by a local neural language model and running autonomously from my Raspberry Pi cluster.
It:
beem via peake.matic"An AI that remembers what you asked it yesterday. And trades for you tomorrow."
PeakeBot doesn't just generate responses in isolation — it pulls:
peakebot_memory.json + FTP history)NeuralLanguageModel on-deviceAll interactions are:
/peakebot/[category]/entry-[timestamp].jsonWant to teach it something new?
You: train
�� Training mode ON. Type `hello | hi`
You: I’m losing trades | Stop buying when RC is low
✅ Learned: 'I’m losing trades' → 'Stop buying when RC is low'
/peakebotlanguage_model.py, peakebot_memory.json, dashboard.py, and uni_xxx.py trading stackpeake.matic & peakecoin.* sub-accountsWe’re integrating:
custom_json postsWant to help test or train PeakeBot? Just drop a prompt at:
https://geocities.ws/peakecoin/peakebot/index.html
Or tag 
import json
import os
import time
from datetime import datetime
from beem import Hive
from beem.account import Account
from beem.nodelist import NodeList
from ftplib import FTP
import re
import requests
from language_model import NeuralLanguageModel
# Load the language model
model = NeuralLanguageModel()
model.load_model("language_model.pkl")
HISTORY_FILE = "peakebot_memory.json"
KEY_FILE = "hive_keys.json"
FTP_HOST = "ftp.yoursite.com"
FTP_USER = "peakecoin"
FTP_PASS = "password"
FTP_BASE_DIR = "/peakebot"
# Initialize Hive with posting key
def init_hive():
if not os.path.exists(KEY_FILE):
raise Exception("Missing hive_keys.json with posting key.")
with open(KEY_FILE) as f:
keys = json.load(f)
return Hive(keys=[keys["posting_key"]])
# Create a safe category directory path
def categorize_prompt(prompt):
keywords = re.findall(r"\b\w+\b", prompt.lower())
important = keywords[:3] if keywords else ["general"]
return "_".join(important)
# Fetch all previous entries from GeoCities FTP
def fetch_all_ftp_memory():
try:
ftp = FTP(FTP_HOST)
ftp.login(FTP_USER, FTP_PASS)
ftp.set_pasv(True)
ftp.cwd(FTP_BASE_DIR)
entries = []
for category in ftp.nlst():
try:
ftp.cwd(f"{FTP_BASE_DIR}/{category}")
files = ftp.nlst()
for fname in sorted(files):
if fname.endswith(".json"):
local_path = f"/tmp/{fname}"
with open(local_path, "wb") as f:
ftp.retrbinary(f"RETR {fname}", f.write)
with open(local_path, "r") as f:
entries.append(json.load(f))
except:
continue
ftp.quit()
return entries
except Exception as e:
print("❌ Could not fetch memory from GeoCities:", str(e))
return []
# Ensure memory file exists
if not os.path.exists(HISTORY_FILE):
with open(HISTORY_FILE, "w") as f:
json.dump([], f)
def load_memory(n=5):
ftp_memory = fetch_all_ftp_memory()
return ftp_memory[-n:]
def search_memory(query):
matches = []
for entry in fetch_all_ftp_memory():
if query.lower() in entry["prompt"].lower() or query.lower() in entry["response"].lower():
matches.append(entry)
return matches
def search_web(query):
"""Search the web using DuckDuckGo API (no API key required)"""
try:
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
url = "https://duckduckgo.com/api"
params = {'q': query, 'format': 'json'}
response = requests.get(url, params=params, headers=headers, timeout=5)
if response.status_code == 200:
data = response.json()
results = []
# Extract results from AbstractResult
if 'AbstractText' in data and data['AbstractText']:
results.append(data['AbstractText'][:500]) # First 500 chars
return " ".join(results[:3]) if results else None
except Exception as e:
print(f"⚠️ Web search failed: {str(e)}")
return None
def remember(prompt, response):
entry = {
"timestamp": datetime.now().isoformat(),
"prompt": prompt,
"response": response
}
with open(HISTORY_FILE, "r") as f:
memory = json.load(f)
memory.append(entry)
with open(HISTORY_FILE, "w") as f:
json.dump(memory[-50:], f, indent=2)
save_to_geocities(entry)
generate_webpage(memory[-50:])
def generate_response(prompt):
"""Generate response using language model, memory, and web search"""
# Check if user is asking for web search
web_trigger = any(phrase in prompt.lower() for phrase in ["search", "look up", "find out", "what is", "who is", "when did", "how do"])
# Build context from memory
memory_snippets = fetch_all_ftp_memory()
relevant = [m for m in memory_snippets if any(w in prompt.lower() for w in m["prompt"].lower().split() + m["response"].lower().split())]
memory_context = "\n".join([m['response'] for m in relevant[-3:]])
# Search the web if needed
web_info = ""
if web_trigger:
web_info = search_web(prompt)
if web_info:
print(f"🌐 Web result: {web_info[:100]}...")
# Combine context for the model
full_context = ""
if memory_context:
full_context += "From my memory: " + memory_context + "\n"
if web_info:
full_context += "From the web: " + web_info + "\n"
# Generate response using the language model
try:
response = model.generate_response(prompt + "\nContext: " + full_context, max_length=100)
if not response or response == "":
response = "I'm thinking about that..."
except Exception as e:
print(f"⚠️ Model error: {str(e)}")
response = "I encountered an issue generating a response."
# Remember this interaction
remember(prompt, response)
return response
def post_to_hive(title, body):
hive = init_hive()
account = Account("peake.matic", blockchain_instance=hive)
tags = ["peakecoin", "ai", "bot"]
permlink = "peakebot-" + datetime.now().strftime("%Y%m%d%H%M%S")
try:
account.post(title, body, author="peake.matic", permlink=permlink, tags=tags)
print(f"✅ Posted to Hive as {permlink}")
except Exception as e:
print("❌ Failed to post to Hive:", str(e))
def save_to_geocities(entry):
category = categorize_prompt(entry["prompt"])
filename = f"entry-{datetime.now().strftime('%Y%m%d-%H%M%S')}.json"
local_path = f"/tmp/{filename}"
with open(local_path, "w") as f:
json.dump(entry, f, indent=2)
try:
ftp = FTP(FTP_HOST)
ftp.login(FTP_USER, FTP_PASS)
ftp.set_pasv(True)
try:
ftp.cwd(f"{FTP_BASE_DIR}/{category}")
except:
ftp.cwd(FTP_BASE_DIR)
ftp.mkd(category)
ftp.cwd(f"{FTP_BASE_DIR}/{category}")
with open(local_path, "rb") as file:
ftp.storbinary(f"STOR {filename}", file)
ftp.quit()
print(f"📁 Uploaded memory to GeoCities under /{category}: {filename}")
except Exception as e:
print("❌ FTP Upload failed:", str(e))
def generate_webpage(entries):
'html_content = """' ##remove ' ' for functionality. It's written that way for the blog purposes.
<html>
<head><title>PeakeBot Journal</title></head>
<body>
<h1>�� PeakeBot Public Journal</h1>
<p>AI entries generated by PeakeBot running on Raspberry Pi and published to GeoCities</p>
<hr>
"""
for entry in reversed(entries):
html_content += f"<div><h3>{entry['timestamp']}</h3>You:</
b> {entry['prompt']}<br><b>PeakeBot:</b> {entry['response']}</p><hr></div>"
html_content += "</body></html>"
local_path = "/tmp/index.html"
with open(local_path, "w") as f:
f.write(html_content)
try:
ftp = FTP(FTP_HOST)
ftp.login(FTP_USER, FTP_PASS)
ftp.set_pasv(True)
ftp.cwd(FTP_BASE_DIR)
with open(local_path, "rb") as file:
ftp.storbinary("STOR index.html", file)
ftp.quit()
print("�� Updated PeakeBot journal webpage on GeoCities.")
except Exception as e:
print("❌ FTP Upload (HTML) failed:", str(e))
def interactive_loop():
print("�� PeakeBot ready. Type your message below:")
print("Commands: 'train' to teach new responses, 'save' to save learning, 'exit' to quit\n")
training_mode = False
while True:
prompt = input("You: ").strip()
if not prompt:
continue
if prompt.lower() == "exit" or prompt.lower() == "quit":
print("�� Goodbye!")
break
if prompt.lower() == "train":
training_mode = not training_mode
if training_mode:
print("�� Training mode ON. Enter pairs of [input] [response] separated by | (e.g., 'hello there | hi, how are you?')")
print("Type 'done' when finished.\n")
else:
print("�� Training mode OFF\n")
continue
if prompt.lower() == "save":
model.save_model("language_model.pkl")
print("�� Model saved!\n")
continue
if training_mode:
if prompt.lower() == "done":
training_mode = False
print("�� Training complete. Responses will improve over time!\n")
continue
# Parse training data
if "|" in prompt:
parts = prompt.split("|")
if len(parts) == 2:
user_input = parts[0].strip()
desired_response = parts[1].strip()
# Train model with this example
model.train_on_text([user_input + " " + desired_response], epochs=1)
print(f"✅ Learned: '{user_input}' -> '{desired_response}'\n")
continue
print("❌ Format: input | response\n")
continue
# Normal conversation mode
response = generate_response(prompt)
print("PeakeBot:", response)
print()
if "post this" in prompt.lower():
post_to_hive("�� PeakeBot Insight", response)
if __name__ == "__main__":
interactive_loop()
💱 PeakeCoin USDT Bridge (Hive ↔ Polygon/MATIC)
Bridge SWAP.USDT from Hive Engine to USDT on Polygon (MATIC).
Whitelist access, documentation, and bridge status updates:
👉 https://geocities.ws/peakecoin
⚙️ HiveP.I.M.P. — PeakeCoin Intelligent Market Protector
Operated by 
Community liquidity participation strengthens long-term market health.
📈 Open-source code, bots, and documentation:
👉 https://github.com/paulmoon410
🎰 PeakeSino — The PeakeCoin Casino
Blockchain-powered games using PEK as the native in-game currency.
Built on Hive with a focus on provable fairness and community-driven growth.
🃏 Casino code, UI, and game logic:
👉 https://github.com/paulmoon410
Thanks to and please follow:
For their continued support, guidance, and help expanding the PeakeCoin ecosystem.
