feat: todo API server, daily brief gen, extract todos, balance checker
- Todo API server (:9001) with CRUD + priority support - Daily brief aggregator (trading + inspiration) - Todo extraction from inspiration digest markdown - DeepSeek balance checker for nav page
This commit is contained in:
Beast
111
scripts/daily_brief_gen.py
Executable file
111
scripts/daily_brief_gen.py
Executable file
@ -0,0 +1,111 @@
|
||||
"""Generate daily brief JSON for nav page - combines trading + inspiration data."""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
from datetime import datetime, timezone, timedelta
|
||||
|
||||
OUTPUT = "/var/www/nav/data/daily_brief.json"
|
||||
BRIEF_DIR = os.path.expanduser("~/freqtrade/user_data/daily_briefs")
|
||||
DIGEST_DIR = os.path.expanduser("~/inspiration-collector/ai-insights/daily")
|
||||
tz = timezone(timedelta(hours=8))
|
||||
today = datetime.now(tz).strftime("%Y-%m-%d")
|
||||
|
||||
|
||||
def get_latest_file(directory, suffix=""):
|
||||
if not os.path.isdir(directory):
|
||||
return None
|
||||
files = [f for f in os.listdir(directory) if f.endswith(suffix)]
|
||||
if not files:
|
||||
return None
|
||||
files.sort(reverse=True)
|
||||
return os.path.join(directory, files[0])
|
||||
|
||||
|
||||
def parse_trading_brief(filepath):
|
||||
"""Extract key stats from trading daily brief."""
|
||||
if not filepath:
|
||||
return {}
|
||||
with open(filepath) as f:
|
||||
content = f.read()
|
||||
|
||||
result = {"has_data": True}
|
||||
# Extract open trades
|
||||
m = re.search(r"当前持仓: (\d+) 笔", content)
|
||||
if m:
|
||||
result["open_trades"] = int(m.group(1))
|
||||
|
||||
m = re.search(r"今日新开: (\d+) 笔", content)
|
||||
if m:
|
||||
result["new_trades"] = int(m.group(1))
|
||||
|
||||
m = re.search(r"今日平仓: (\d+) 笔", content)
|
||||
if m:
|
||||
result["closed_trades"] = int(m.group(1))
|
||||
|
||||
# Extract strategy name
|
||||
m = re.search(r"策略: (\S+) \|", content)
|
||||
if m:
|
||||
result["strategy"] = m.group(1)
|
||||
|
||||
# Extract health
|
||||
m = re.search(r"健康度: (\S+) \((\d+/\d+)\)", content)
|
||||
if m:
|
||||
result["health"] = m.group(1)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def parse_digest(filepath):
|
||||
"""Extract summary and categories from inspiration digest."""
|
||||
if not filepath:
|
||||
return {}
|
||||
with open(filepath) as f:
|
||||
content = f.read()
|
||||
|
||||
result = {"has_data": True}
|
||||
# Extract date
|
||||
m = re.search(r"date: (\d{4}-\d{2}-\d{2})", content)
|
||||
if m:
|
||||
result["date"] = m.group(1)
|
||||
|
||||
# Extract summary line (first line after the date heading)
|
||||
lines = content.split("\n")
|
||||
for i, line in enumerate(lines):
|
||||
if line.startswith("#") and i + 1 < len(lines):
|
||||
next_line = lines[i + 1].strip()
|
||||
if next_line:
|
||||
result["summary"] = next_line[:100]
|
||||
break
|
||||
|
||||
# Extract categories and todo count
|
||||
cats = re.findall(r"\*\*(.+?)\*\*：(\d+)条", content)
|
||||
if cats:
|
||||
result["categories"] = {c[0]: int(c[1]) for c in cats}
|
||||
|
||||
todos = re.findall(r"\[ \] (.+)", content)
|
||||
result["todo_count"] = len(todos)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
latest_brief = get_latest_file(BRIEF_DIR, ".txt")
|
||||
latest_digest = get_latest_file(DIGEST_DIR, "_digest.md")
|
||||
|
||||
trading = parse_trading_brief(latest_brief)
|
||||
inspiration = parse_digest(latest_digest)
|
||||
|
||||
brief = {
|
||||
"date": today,
|
||||
"trading": trading,
|
||||
"inspiration": inspiration,
|
||||
"has_trading": bool(trading.get("has_data")),
|
||||
"has_inspiration": bool(inspiration.get("has_data")),
|
||||
"generated_at": datetime.now(tz).strftime("%Y-%m-%d %H:%M"),
|
||||
}
|
||||
|
||||
os.makedirs(os.path.dirname(OUTPUT), exist_ok=True)
|
||||
with open(OUTPUT, "w") as f:
|
||||
json.dump(brief, f, ensure_ascii=False, indent=2)
|
||||
|
||||
print(f"Brief written: {OUTPUT}")
|
||||
63
scripts/extract_todos.py
Executable file
63
scripts/extract_todos.py
Executable file
@ -0,0 +1,63 @@
|
||||
"""Extract todos from the latest inspiration digest + sync to todo API."""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import requests
|
||||
|
||||
DIGEST_DIR = os.path.expanduser("~/inspiration-collector/ai-insights/daily")
|
||||
TODO_API = "http://127.0.0.1:9001/api/todos"
|
||||
|
||||
|
||||
def get_latest_digest():
|
||||
"""Find the most recent digest file."""
|
||||
if not os.path.isdir(DIGEST_DIR):
|
||||
return None
|
||||
files = [f for f in os.listdir(DIGEST_DIR) if f.endswith("_digest.md")]
|
||||
if not files:
|
||||
return None
|
||||
files.sort(reverse=True)
|
||||
return os.path.join(DIGEST_DIR, files[0])
|
||||
|
||||
|
||||
def extract_todos(filepath):
|
||||
"""Extract [ ] todo items from the digest markdown."""
|
||||
with open(filepath) as f:
|
||||
content = f.read()
|
||||
|
||||
todos = []
|
||||
# Match markdown checkboxes: - [ ] something
|
||||
for line in content.split("\n"):
|
||||
m = re.match(r"\s*[-*]\s*\[\s*[ ]\s*\]\s*(.+)", line)
|
||||
if m:
|
||||
todos.append(m.group(1).strip())
|
||||
return todos
|
||||
|
||||
|
||||
def push_todos(todos):
|
||||
"""Send extracted todos to the API (skip if already exists)."""
|
||||
# Get existing todos
|
||||
try:
|
||||
resp = requests.get(TODO_API, timeout=5)
|
||||
existing = [t["text"] for t in resp.json().get("todos", [])]
|
||||
except:
|
||||
existing = []
|
||||
|
||||
count = 0
|
||||
for text in todos:
|
||||
if text not in existing:
|
||||
requests.post(TODO_API, json={"text": text, "source": "inspiration"}, timeout=5)
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
filepath = get_latest_digest()
|
||||
if not filepath:
|
||||
print("No digest found")
|
||||
exit(0)
|
||||
|
||||
todos = extract_todos(filepath)
|
||||
print(f"Found {len(todos)} todos in {os.path.basename(filepath)}")
|
||||
if todos:
|
||||
pushed = push_todos(todos)
|
||||
print(f"Pushed {pushed} new todos to API")
|
||||
106
scripts/todo_server.py
Executable file
106
scripts/todo_server.py
Executable file
@ -0,0 +1,106 @@
|
||||
"""Todo API server - lightweight todo management for nav page.
|
||||
Runs on port 9001, managed by Caddy proxy at /api/todos/*
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import uuid
|
||||
from datetime import datetime, timezone, timedelta
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
|
||||
TODOS_FILE = "/var/www/nav/data/todos.json"
|
||||
tz = timezone(timedelta(hours=8))
|
||||
|
||||
|
||||
def load_todos():
|
||||
if os.path.exists(TODOS_FILE):
|
||||
with open(TODOS_FILE) as f:
|
||||
return json.load(f)
|
||||
return []
|
||||
|
||||
|
||||
def save_todos(todos):
|
||||
os.makedirs(os.path.dirname(TODOS_FILE), exist_ok=True)
|
||||
with open(TODOS_FILE, "w") as f:
|
||||
json.dump(todos, f, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
class TodoHandler(BaseHTTPRequestHandler):
|
||||
def _send(self, data, status=200):
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||||
self.send_header("Access-Control-Allow-Origin", "*")
|
||||
self.send_header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS")
|
||||
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
||||
self.end_headers()
|
||||
self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
|
||||
|
||||
def do_OPTIONS(self):
|
||||
self._send({})
|
||||
|
||||
def do_GET(self):
|
||||
if self.path == "/api/todos":
|
||||
todos = load_todos()
|
||||
self._send({"todos": todos})
|
||||
else:
|
||||
self._send({"error": "not found"}, 404)
|
||||
|
||||
def do_POST(self):
|
||||
if self.path == "/api/todos":
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
body = json.loads(self.rfile.read(length).decode("utf-8"))
|
||||
todos = load_todos()
|
||||
now = datetime.now(tz).strftime("%Y-%m-%d %H:%M")
|
||||
todo = {
|
||||
"id": str(uuid.uuid4())[:8],
|
||||
"text": body.get("text", ""),
|
||||
"done": False,
|
||||
"source": body.get("source", "manual"),
|
||||
"priority": body.get("priority", "medium"),
|
||||
"created_at": now,
|
||||
}
|
||||
todos.insert(0, todo)
|
||||
save_todos(todos)
|
||||
self._send({"todo": todo}, 201)
|
||||
else:
|
||||
self._send({"error": "not found"}, 404)
|
||||
|
||||
def do_PUT(self):
|
||||
parts = self.path.split("/")
|
||||
if len(parts) == 4 and parts[1] == "api" and parts[2] == "todos":
|
||||
tid = parts[3]
|
||||
length = int(self.headers.get("Content-Length", 0))
|
||||
body = json.loads(self.rfile.read(length).decode("utf-8"))
|
||||
todos = load_todos()
|
||||
for t in todos:
|
||||
if t["id"] == tid:
|
||||
if "done" in body:
|
||||
t["done"] = body["done"]
|
||||
if "text" in body:
|
||||
t["text"] = body["text"]
|
||||
save_todos(todos)
|
||||
self._send({"todo": t})
|
||||
return
|
||||
self._send({"error": "not found"}, 404)
|
||||
else:
|
||||
self._send({"error": "not found"}, 404)
|
||||
|
||||
def do_DELETE(self):
|
||||
parts = self.path.split("/")
|
||||
if len(parts) == 4 and parts[1] == "api" and parts[2] == "todos":
|
||||
tid = parts[3]
|
||||
todos = load_todos()
|
||||
todos = [t for t in todos if t["id"] != tid]
|
||||
save_todos(todos)
|
||||
self._send({"deleted": tid})
|
||||
else:
|
||||
self._send({"error": "not found"}, 404)
|
||||
|
||||
def log_message(self, format, *args):
|
||||
pass # quiet
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
port = int(os.environ.get("PORT", 9001))
|
||||
server = HTTPServer(("127.0.0.1", port), TodoHandler)
|
||||
print(f"Todo API running on :{port}")
|
||||
server.serve_forever()
|
||||
Reference in New Issue
Block a user