745ce62322
- 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
112 lines
3.2 KiB
Python
Executable File
112 lines
3.2 KiB
Python
Executable File
"""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}")
|