1613 lines
67 KiB
Python
1613 lines
67 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class ManagedConnection(sqlite3.Connection):
|
|
"""Commit or roll back, then release the SQLite file handle on context exit."""
|
|
|
|
def __exit__(self, exc_type, exc_value, traceback):
|
|
try:
|
|
return super().__exit__(exc_type, exc_value, traceback)
|
|
finally:
|
|
self.close()
|
|
|
|
|
|
class ReviewDatabase:
|
|
def __init__(self, path: Path) -> None:
|
|
self.path = path
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._initialize()
|
|
|
|
def connect(self) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(self.path, timeout=20, factory=ManagedConnection)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
return connection
|
|
|
|
def _initialize(self) -> None:
|
|
with self.connect() as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
|
password_salt TEXT NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_sessions (
|
|
token_hash TEXT PRIMARY KEY,
|
|
user_id INTEGER NOT NULL,
|
|
csrf_token TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
last_seen_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_user_sessions_user
|
|
ON user_sessions(user_id, expires_at);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_credentials (
|
|
user_id INTEGER PRIMARY KEY,
|
|
encrypted_payload TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS user_birth_profiles (
|
|
user_id INTEGER PRIMARY KEY,
|
|
encrypted_payload TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS system_settings (
|
|
setting_key TEXT PRIMARY KEY,
|
|
encrypted_payload TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS llm_usage (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
feature TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
model TEXT NOT NULL DEFAULT '',
|
|
status TEXT NOT NULL,
|
|
latency_ms INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_llm_usage_user_time
|
|
ON llm_usage(user_id, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS dashboard_snapshots (
|
|
trade_date TEXT PRIMARY KEY,
|
|
source TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
record_count INTEGER NOT NULL DEFAULT 0,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS sync_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
trade_date TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
status TEXT NOT NULL,
|
|
started_at TEXT NOT NULL,
|
|
finished_at TEXT,
|
|
record_count INTEGER NOT NULL DEFAULT 0,
|
|
message TEXT NOT NULL DEFAULT ''
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_sync_runs_trade_date
|
|
ON sync_runs(trade_date, id DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS data_snapshots (
|
|
kind TEXT NOT NULL,
|
|
cache_key TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
payload TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (kind, cache_key)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS watchlist (
|
|
user_id INTEGER NOT NULL,
|
|
code TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
sector TEXT NOT NULL DEFAULT '',
|
|
color TEXT NOT NULL DEFAULT 'red',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (user_id, code),
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS review_notes (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
code TEXT NOT NULL DEFAULT '',
|
|
stock_name TEXT NOT NULL DEFAULT '',
|
|
trade_date TEXT NOT NULL,
|
|
content TEXT NOT NULL DEFAULT '',
|
|
plan TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_review_notes_code_date
|
|
ON review_notes(code, trade_date DESC, id DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS reason_overrides (
|
|
trade_date TEXT NOT NULL,
|
|
code TEXT NOT NULL,
|
|
reason TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (trade_date, code)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS seat_aliases (
|
|
seat_name TEXT PRIMARY KEY,
|
|
alias TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS sector_phase_overrides (
|
|
name TEXT PRIMARY KEY,
|
|
element TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS stock_master (
|
|
ts_code TEXT PRIMARY KEY,
|
|
code TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
industry TEXT NOT NULL DEFAULT '',
|
|
market TEXT NOT NULL DEFAULT '',
|
|
list_date TEXT NOT NULL DEFAULT '',
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_stock_master_code ON stock_master(code);
|
|
|
|
CREATE TABLE IF NOT EXISTS daily_bars (
|
|
trade_date TEXT NOT NULL,
|
|
ts_code TEXT NOT NULL,
|
|
open REAL NOT NULL DEFAULT 0,
|
|
high REAL NOT NULL DEFAULT 0,
|
|
low REAL NOT NULL DEFAULT 0,
|
|
close REAL NOT NULL DEFAULT 0,
|
|
pct_chg REAL NOT NULL DEFAULT 0,
|
|
vol REAL NOT NULL DEFAULT 0,
|
|
amount REAL NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (trade_date, ts_code)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_daily_bars_code_date
|
|
ON daily_bars(ts_code, trade_date DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS daily_indicators (
|
|
trade_date TEXT NOT NULL,
|
|
ts_code TEXT NOT NULL,
|
|
turnover_rate REAL NOT NULL DEFAULT 0,
|
|
volume_ratio REAL NOT NULL DEFAULT 0,
|
|
total_mv REAL NOT NULL DEFAULT 0,
|
|
circ_mv REAL NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (trade_date, ts_code)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS moneyflow_daily (
|
|
trade_date TEXT NOT NULL,
|
|
ts_code TEXT NOT NULL,
|
|
net_mf_amount REAL NOT NULL DEFAULT 0,
|
|
large_net_amount REAL NOT NULL DEFAULT 0,
|
|
medium_net_amount REAL NOT NULL DEFAULT 0,
|
|
small_net_amount REAL NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (trade_date, ts_code)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS screener_strategies (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER,
|
|
name TEXT NOT NULL,
|
|
description TEXT NOT NULL DEFAULT '',
|
|
regimes TEXT NOT NULL,
|
|
formula TEXT NOT NULL,
|
|
builtin INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS screener_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER,
|
|
trade_date TEXT NOT NULL,
|
|
regime TEXT NOT NULL,
|
|
strategy_name TEXT NOT NULL,
|
|
formula TEXT NOT NULL,
|
|
result TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS mentor_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
mentor_id TEXT NOT NULL,
|
|
trade_date TEXT NOT NULL,
|
|
role TEXT NOT NULL,
|
|
content TEXT NOT NULL,
|
|
meta TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_mentor_messages_conversation
|
|
ON mentor_messages(user_id, mentor_id, trade_date, id DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS strategy_tracks (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
run_id INTEGER NOT NULL,
|
|
selection_date TEXT NOT NULL,
|
|
strategy_name TEXT NOT NULL,
|
|
ts_code TEXT NOT NULL,
|
|
code TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
sector TEXT NOT NULL DEFAULT '',
|
|
entry_price REAL NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
UNIQUE(user_id, run_id, ts_code),
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
|
|
FOREIGN KEY (run_id) REFERENCES screener_runs(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_strategy_tracks_user_run
|
|
ON strategy_tracks(user_id, run_id DESC, id);
|
|
|
|
CREATE TABLE IF NOT EXISTS alerts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
kind TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
content TEXT NOT NULL DEFAULT '',
|
|
available_date TEXT NOT NULL,
|
|
code TEXT NOT NULL DEFAULT '',
|
|
dedupe_key TEXT NOT NULL,
|
|
is_read INTEGER NOT NULL DEFAULT 0,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
read_at TEXT,
|
|
UNIQUE(user_id, dedupe_key),
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_alerts_user_due
|
|
ON alerts(user_id, available_date, is_read, id DESC);
|
|
"""
|
|
)
|
|
user_columns = {
|
|
str(row["name"]) for row in connection.execute("PRAGMA table_info(users)")
|
|
}
|
|
migrations = {
|
|
"role": "ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'user'",
|
|
"llm_mode": "ALTER TABLE users ADD COLUMN llm_mode TEXT NOT NULL DEFAULT 'auto'",
|
|
"membership_status": "ALTER TABLE users ADD COLUMN membership_status TEXT NOT NULL DEFAULT 'inactive'",
|
|
"membership_plan": "ALTER TABLE users ADD COLUMN membership_plan TEXT NOT NULL DEFAULT ''",
|
|
"membership_starts_at": "ALTER TABLE users ADD COLUMN membership_starts_at TEXT",
|
|
"membership_expires_at": "ALTER TABLE users ADD COLUMN membership_expires_at TEXT",
|
|
}
|
|
for column, statement in migrations.items():
|
|
if column not in user_columns:
|
|
connection.execute(statement)
|
|
connection.execute(
|
|
"""
|
|
UPDATE users SET role = 'admin'
|
|
WHERE id = (SELECT MIN(id) FROM users)
|
|
AND NOT EXISTS (SELECT 1 FROM users WHERE role = 'admin')
|
|
"""
|
|
)
|
|
watchlist_columns = {
|
|
str(row["name"]) for row in connection.execute("PRAGMA table_info(watchlist)")
|
|
}
|
|
if "user_id" not in watchlist_columns:
|
|
connection.execute("ALTER TABLE watchlist RENAME TO watchlist_legacy")
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE watchlist (
|
|
user_id INTEGER NOT NULL,
|
|
code TEXT NOT NULL,
|
|
name TEXT NOT NULL,
|
|
sector TEXT NOT NULL DEFAULT '',
|
|
color TEXT NOT NULL DEFAULT 'red',
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (user_id, code),
|
|
FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
|
|
)
|
|
"""
|
|
)
|
|
first_user = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
|
|
if first_user and first_user["id"]:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO watchlist
|
|
(user_id, code, name, sector, color, created_at, updated_at)
|
|
SELECT ?, code, name, sector, color, created_at, updated_at
|
|
FROM watchlist_legacy
|
|
""",
|
|
(int(first_user["id"]),),
|
|
)
|
|
connection.execute("DROP TABLE watchlist_legacy")
|
|
note_columns = {
|
|
str(row["name"]) for row in connection.execute("PRAGMA table_info(review_notes)")
|
|
}
|
|
if "user_id" not in note_columns:
|
|
connection.execute("ALTER TABLE review_notes ADD COLUMN user_id INTEGER")
|
|
first_user = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
|
|
if first_user and first_user["id"]:
|
|
connection.execute(
|
|
"UPDATE review_notes SET user_id = ? WHERE user_id IS NULL",
|
|
(int(first_user["id"]),),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_review_notes_user_date
|
|
ON review_notes(user_id, trade_date DESC, id DESC)
|
|
"""
|
|
)
|
|
strategy_columns = {
|
|
str(row["name"]) for row in connection.execute("PRAGMA table_info(screener_strategies)")
|
|
}
|
|
if "user_id" not in strategy_columns:
|
|
connection.execute("ALTER TABLE screener_strategies ADD COLUMN user_id INTEGER")
|
|
run_columns = {
|
|
str(row["name"]) for row in connection.execute("PRAGMA table_info(screener_runs)")
|
|
}
|
|
if "user_id" not in run_columns:
|
|
connection.execute("ALTER TABLE screener_runs ADD COLUMN user_id INTEGER")
|
|
if first_user and first_user["id"]:
|
|
first_user_id = int(first_user["id"])
|
|
connection.execute(
|
|
"UPDATE screener_strategies SET user_id = ? WHERE builtin = 0 AND user_id IS NULL",
|
|
(first_user_id,),
|
|
)
|
|
connection.execute(
|
|
"UPDATE screener_runs SET user_id = ? WHERE user_id IS NULL",
|
|
(first_user_id,),
|
|
)
|
|
connection.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_screener_strategies_user
|
|
ON screener_strategies(user_id, builtin, updated_at DESC)
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"""
|
|
CREATE INDEX IF NOT EXISTS idx_screener_runs_user_date
|
|
ON screener_runs(user_id, trade_date DESC, id DESC)
|
|
"""
|
|
)
|
|
|
|
def count_users(self) -> int:
|
|
with self.connect() as connection:
|
|
row = connection.execute("SELECT COUNT(*) AS total FROM users").fetchone()
|
|
return int(row["total"] if row else 0)
|
|
|
|
def first_user_id(self) -> int:
|
|
with self.connect() as connection:
|
|
row = connection.execute("SELECT MIN(id) AS id FROM users").fetchone()
|
|
return int(row["id"] or 0) if row else 0
|
|
|
|
def create_user(
|
|
self,
|
|
username: str,
|
|
password_salt: str,
|
|
password_hash: str,
|
|
) -> dict[str, Any]:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
try:
|
|
with self.connect() as connection:
|
|
role = "admin" if int(connection.execute("SELECT COUNT(*) FROM users").fetchone()[0]) == 0 else "user"
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO users
|
|
(username, password_salt, password_hash, role, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(username, password_salt, password_hash, role, now, now),
|
|
)
|
|
user_id = int(cursor.lastrowid)
|
|
except sqlite3.IntegrityError as exc:
|
|
raise ValueError("该账号名已被使用。") from exc
|
|
return {"id": user_id, "username": username, "role": role, "created_at": now}
|
|
|
|
def user_by_username(self, username: str) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT id, username, password_salt, password_hash, role, llm_mode,
|
|
membership_status, membership_plan, membership_starts_at,
|
|
membership_expires_at, created_at
|
|
FROM users WHERE username = ? COLLATE NOCASE
|
|
""",
|
|
(username,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def user_password(self, user_id: int) -> dict[str, str] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT password_salt, password_hash FROM users WHERE id = ?",
|
|
(user_id,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def update_user_password(self, user_id: int, password_salt: str, password_hash: str) -> bool:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"UPDATE users SET password_salt = ?, password_hash = ?, updated_at = ? WHERE id = ?",
|
|
(password_salt, password_hash, now, user_id),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def delete_user(self, user_id: int) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
|
|
return cursor.rowcount > 0
|
|
|
|
def create_session(
|
|
self,
|
|
session_hash: str,
|
|
user_id: int,
|
|
csrf_token: str,
|
|
expires_at: str,
|
|
) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute("DELETE FROM user_sessions WHERE expires_at <= ?", (now,))
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO user_sessions
|
|
(token_hash, user_id, csrf_token, expires_at, created_at, last_seen_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(session_hash, user_id, csrf_token, expires_at, now, now),
|
|
)
|
|
|
|
def session_user(self, session_hash: str) -> dict[str, Any] | None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT u.id, u.username, u.role, u.llm_mode, u.membership_status,
|
|
u.membership_plan, u.membership_starts_at, u.membership_expires_at,
|
|
u.created_at, s.csrf_token, s.expires_at
|
|
FROM user_sessions AS s
|
|
JOIN users AS u ON u.id = s.user_id
|
|
WHERE s.token_hash = ? AND s.expires_at > ?
|
|
""",
|
|
(session_hash, now),
|
|
).fetchone()
|
|
if row:
|
|
connection.execute(
|
|
"UPDATE user_sessions SET last_seen_at = ? WHERE token_hash = ?",
|
|
(now, session_hash),
|
|
)
|
|
return dict(row) if row else None
|
|
|
|
def delete_session(self, session_hash: str) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM user_sessions WHERE token_hash = ?",
|
|
(session_hash,),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def get_user_credentials(self, user_id: int) -> str:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT encrypted_payload FROM user_credentials WHERE user_id = ?",
|
|
(user_id,),
|
|
).fetchone()
|
|
return str(row["encrypted_payload"]) if row else ""
|
|
|
|
def save_user_credentials(self, user_id: int, encrypted_payload: str) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO user_credentials (user_id, encrypted_payload, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(user_id) DO UPDATE SET
|
|
encrypted_payload = excluded.encrypted_payload,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(user_id, encrypted_payload, now),
|
|
)
|
|
|
|
def list_user_credentials(self) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT user_id, encrypted_payload FROM user_credentials ORDER BY user_id"
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def get_system_setting(self, key: str) -> str:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT encrypted_payload FROM system_settings WHERE setting_key = ?",
|
|
(key,),
|
|
).fetchone()
|
|
return str(row["encrypted_payload"]) if row else ""
|
|
|
|
def save_system_setting(self, key: str, encrypted_payload: str) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO system_settings (setting_key, encrypted_payload, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(setting_key) DO UPDATE SET
|
|
encrypted_payload = excluded.encrypted_payload,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(key, encrypted_payload, now),
|
|
)
|
|
|
|
def user_access(self, user_id: int) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT id, username, role, llm_mode, membership_status, membership_plan,
|
|
membership_starts_at, membership_expires_at, created_at
|
|
FROM users WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def list_users(self) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT id, username, role, llm_mode, membership_status, membership_plan,
|
|
membership_starts_at, membership_expires_at, created_at
|
|
FROM users ORDER BY id
|
|
"""
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def update_user_llm_mode(self, user_id: int, mode: str) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"UPDATE users SET llm_mode = ?, updated_at = ? WHERE id = ?",
|
|
(mode, now, user_id),
|
|
)
|
|
|
|
def update_membership(
|
|
self,
|
|
user_id: int,
|
|
status: str,
|
|
plan: str,
|
|
starts_at: str | None,
|
|
expires_at: str | None,
|
|
) -> bool:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE users
|
|
SET membership_status = ?, membership_plan = ?,
|
|
membership_starts_at = ?, membership_expires_at = ?, updated_at = ?
|
|
WHERE id = ?
|
|
""",
|
|
(status, plan, starts_at, expires_at, now, user_id),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def record_llm_usage(
|
|
self,
|
|
user_id: int,
|
|
feature: str,
|
|
source: str,
|
|
model: str,
|
|
status: str,
|
|
latency_ms: int = 0,
|
|
) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO llm_usage
|
|
(user_id, feature, source, model, status, latency_ms, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(user_id, feature, source, model, status, int(latency_ms), now),
|
|
)
|
|
|
|
def count_llm_usage_since(self, user_id: int, source: str, since: str) -> int:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT COUNT(*) AS total FROM llm_usage
|
|
WHERE user_id = ? AND source = ? AND created_at >= ?
|
|
""",
|
|
(user_id, source, since),
|
|
).fetchone()
|
|
return int(row["total"] if row else 0)
|
|
|
|
def get_user_birth_profile(self, user_id: int) -> str:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT encrypted_payload FROM user_birth_profiles WHERE user_id = ?",
|
|
(user_id,),
|
|
).fetchone()
|
|
return str(row["encrypted_payload"]) if row else ""
|
|
|
|
def save_user_birth_profile(self, user_id: int, encrypted_payload: str) -> None:
|
|
now = datetime.now(timezone.utc).isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO user_birth_profiles (user_id, encrypted_payload, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(user_id) DO UPDATE SET
|
|
encrypted_payload = excluded.encrypted_payload,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(user_id, encrypted_payload, now),
|
|
)
|
|
|
|
def delete_user_birth_profile(self, user_id: int) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM user_birth_profiles WHERE user_id = ?",
|
|
(user_id,),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def get_snapshot(self, trade_date: str) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT payload FROM dashboard_snapshots WHERE trade_date = ?",
|
|
(trade_date,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return json.loads(row["payload"])
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def get_latest_real_snapshot(
|
|
self, trade_date: str, strictly_before: bool = False
|
|
) -> dict[str, Any] | None:
|
|
operator = "<" if strictly_before else "<="
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
f"""
|
|
SELECT payload FROM dashboard_snapshots
|
|
WHERE trade_date {operator} ? AND source != 'demo'
|
|
ORDER BY trade_date DESC LIMIT 1
|
|
""",
|
|
(trade_date,),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return json.loads(row["payload"])
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def save_snapshot(self, trade_date: str, source: str, payload: dict[str, Any]) -> None:
|
|
updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
record_count = sum(
|
|
len(payload.get(key) or [])
|
|
for key in ("limits", "broken", "down_limits", "yesterday_limits")
|
|
)
|
|
content = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO dashboard_snapshots
|
|
(trade_date, source, payload, record_count, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(trade_date) DO UPDATE SET
|
|
source = excluded.source,
|
|
payload = excluded.payload,
|
|
record_count = excluded.record_count,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(trade_date, source, content, record_count, updated_at),
|
|
)
|
|
|
|
def get_data_snapshot(self, kind: str, cache_key: str) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT payload FROM data_snapshots WHERE kind = ? AND cache_key = ?",
|
|
(kind, cache_key),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return json.loads(row["payload"])
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def get_latest_data_snapshot(
|
|
self,
|
|
kind: str,
|
|
cache_key_prefix: str,
|
|
maximum_cache_key: str,
|
|
exclude_source: str = "",
|
|
) -> dict[str, Any] | None:
|
|
source_clause = " AND source != ?" if exclude_source else ""
|
|
parameters: list[Any] = [kind, f"{cache_key_prefix}%", maximum_cache_key]
|
|
if exclude_source:
|
|
parameters.append(exclude_source)
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
f"""
|
|
SELECT payload FROM data_snapshots
|
|
WHERE kind = ? AND cache_key LIKE ? AND cache_key <= ?{source_clause}
|
|
ORDER BY cache_key DESC LIMIT 1
|
|
""",
|
|
parameters,
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
return json.loads(row["payload"])
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def save_data_snapshot(
|
|
self, kind: str, cache_key: str, source: str, payload: dict[str, Any]
|
|
) -> None:
|
|
updated_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
content = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO data_snapshots (kind, cache_key, source, payload, updated_at)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
ON CONFLICT(kind, cache_key) DO UPDATE SET
|
|
source = excluded.source,
|
|
payload = excluded.payload,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(kind, cache_key, source, content, updated_at),
|
|
)
|
|
|
|
def list_watchlist(self, user_id: int) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT code, name, sector, color, created_at, updated_at
|
|
FROM watchlist WHERE user_id = ? ORDER BY updated_at DESC, code
|
|
""",
|
|
(int(user_id),),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def save_watchlist(
|
|
self, user_id: int, code: str, name: str, sector: str, color: str
|
|
) -> None:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO watchlist
|
|
(user_id, code, name, sector, color, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(user_id, code) DO UPDATE SET
|
|
name = excluded.name,
|
|
sector = excluded.sector,
|
|
color = excluded.color,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(int(user_id), code, name, sector, color, now, now),
|
|
)
|
|
|
|
def delete_watchlist(self, user_id: int, code: str) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM watchlist WHERE user_id = ? AND code = ?",
|
|
(int(user_id), code),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def list_notes(
|
|
self,
|
|
user_id: int,
|
|
code: str = "",
|
|
trade_date: str = "",
|
|
scope: str = "all",
|
|
) -> list[dict[str, Any]]:
|
|
clauses: list[str] = ["user_id = ?"]
|
|
parameters: list[Any] = [int(user_id)]
|
|
if scope == "daily":
|
|
clauses.append("code = ''")
|
|
elif scope == "stock":
|
|
clauses.append("code <> ''")
|
|
if code:
|
|
clauses.append("code = ?")
|
|
parameters.append(code)
|
|
if trade_date:
|
|
clauses.append("trade_date = ?")
|
|
parameters.append(trade_date)
|
|
where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT id, code, stock_name, trade_date, content, plan, created_at, updated_at
|
|
FROM review_notes {where}
|
|
ORDER BY trade_date DESC, updated_at DESC, id DESC LIMIT 200
|
|
""",
|
|
parameters,
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def save_note(
|
|
self,
|
|
user_id: int,
|
|
code: str,
|
|
stock_name: str,
|
|
trade_date: str,
|
|
content: str,
|
|
plan: str,
|
|
note_id: int | None = None,
|
|
) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
if note_id:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE review_notes
|
|
SET code = ?, stock_name = ?, trade_date = ?, content = ?, plan = ?, updated_at = ?
|
|
WHERE id = ? AND user_id = ?
|
|
""",
|
|
(code, stock_name, trade_date, content, plan, now, note_id, int(user_id)),
|
|
)
|
|
if cursor.rowcount == 0:
|
|
raise ValueError("复盘笔记不存在。")
|
|
return note_id
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO review_notes
|
|
(user_id, code, stock_name, trade_date, content, plan, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(int(user_id), code, stock_name, trade_date, content, plan, now, now),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def delete_note(self, user_id: int, note_id: int) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM review_notes WHERE id = ? AND user_id = ?",
|
|
(note_id, int(user_id)),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def save_reason_override(self, trade_date: str, code: str, reason: str) -> None:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO reason_overrides (trade_date, code, reason, updated_at)
|
|
VALUES (?, ?, ?, ?)
|
|
ON CONFLICT(trade_date, code) DO UPDATE SET
|
|
reason = excluded.reason,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(trade_date, code, reason, now),
|
|
)
|
|
|
|
def reason_overrides(self, trade_date: str) -> dict[str, str]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT code, reason FROM reason_overrides WHERE trade_date = ?",
|
|
(trade_date,),
|
|
).fetchall()
|
|
return {row["code"]: row["reason"] for row in rows}
|
|
|
|
def list_seat_aliases(self) -> dict[str, str]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
|
|
return {row["seat_name"]: row["alias"] for row in rows}
|
|
|
|
def save_seat_alias(self, seat_name: str, alias: str) -> None:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO seat_aliases (seat_name, alias, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(seat_name) DO UPDATE SET
|
|
alias = excluded.alias,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(seat_name, alias, now),
|
|
)
|
|
|
|
def list_sector_phase_overrides(self) -> dict[str, str]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT name, element FROM sector_phase_overrides ORDER BY updated_at DESC, name"
|
|
).fetchall()
|
|
return {row["name"]: row["element"] for row in rows}
|
|
|
|
def save_sector_phase_override(self, name: str, element: str) -> None:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO sector_phase_overrides (name, element, updated_at)
|
|
VALUES (?, ?, ?)
|
|
ON CONFLICT(name) DO UPDATE SET
|
|
element = excluded.element,
|
|
updated_at = excluded.updated_at
|
|
""",
|
|
(name, element, now),
|
|
)
|
|
|
|
def delete_sector_phase_override(self, name: str) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM sector_phase_overrides WHERE name = ?",
|
|
(name,),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def upsert_stock_master(self, rows: list[dict[str, Any]]) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
values = [
|
|
(
|
|
row.get("ts_code", ""),
|
|
str(row.get("ts_code", "")).split(".")[0],
|
|
row.get("name") or "--",
|
|
row.get("industry") or "",
|
|
row.get("market") or "",
|
|
str(row.get("list_date") or ""),
|
|
now,
|
|
)
|
|
for row in rows if row.get("ts_code")
|
|
]
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO stock_master
|
|
(ts_code, code, name, industry, market, list_date, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(ts_code) DO UPDATE SET
|
|
code=excluded.code, name=excluded.name, industry=excluded.industry,
|
|
market=excluded.market, list_date=excluded.list_date, updated_at=excluded.updated_at
|
|
""",
|
|
values,
|
|
)
|
|
return len(values)
|
|
|
|
def search_stock_master(self, query: str, limit: int = 12) -> list[dict[str, Any]]:
|
|
text = str(query or "").strip()
|
|
if not text:
|
|
return []
|
|
escaped = text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT ts_code, code, name, industry, market, list_date
|
|
FROM stock_master
|
|
WHERE code = ? OR name = ? OR name LIKE ? ESCAPE '\\'
|
|
ORDER BY
|
|
CASE WHEN code = ? THEN 0 WHEN name = ? THEN 1 ELSE 2 END,
|
|
list_date DESC,
|
|
code
|
|
LIMIT ?
|
|
""",
|
|
(text, text, f"%{escaped}%", text, text, max(1, min(30, int(limit)))),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def upsert_daily_bars(self, rows: list[dict[str, Any]]) -> int:
|
|
values = [
|
|
(
|
|
str(row.get("trade_date") or ""), row.get("ts_code", ""),
|
|
float(row.get("open") or 0), float(row.get("high") or 0),
|
|
float(row.get("low") or 0), float(row.get("close") or 0),
|
|
float(row.get("pct_chg") or 0), float(row.get("vol") or 0),
|
|
float(row.get("amount") or 0),
|
|
)
|
|
for row in rows if row.get("trade_date") and row.get("ts_code")
|
|
]
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO daily_bars
|
|
(trade_date, ts_code, open, high, low, close, pct_chg, vol, amount)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
|
|
open=excluded.open, high=excluded.high, low=excluded.low,
|
|
close=excluded.close, pct_chg=excluded.pct_chg,
|
|
vol=excluded.vol, amount=excluded.amount
|
|
""",
|
|
values,
|
|
)
|
|
return len(values)
|
|
|
|
def upsert_daily_indicators(self, rows: list[dict[str, Any]]) -> int:
|
|
values = [
|
|
(
|
|
str(row.get("trade_date") or ""), row.get("ts_code", ""),
|
|
float(row.get("turnover_rate") or 0), float(row.get("volume_ratio") or 0),
|
|
float(row.get("total_mv") or 0), float(row.get("circ_mv") or 0),
|
|
)
|
|
for row in rows if row.get("trade_date") and row.get("ts_code")
|
|
]
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO daily_indicators
|
|
(trade_date, ts_code, turnover_rate, volume_ratio, total_mv, circ_mv)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
|
|
turnover_rate=excluded.turnover_rate, volume_ratio=excluded.volume_ratio,
|
|
total_mv=excluded.total_mv, circ_mv=excluded.circ_mv
|
|
""",
|
|
values,
|
|
)
|
|
return len(values)
|
|
|
|
def upsert_moneyflow(self, rows: list[dict[str, Any]]) -> int:
|
|
values = []
|
|
for row in rows:
|
|
if not row.get("trade_date") or not row.get("ts_code"):
|
|
continue
|
|
large_net = (
|
|
float(row.get("buy_lg_amount") or 0) + float(row.get("buy_elg_amount") or 0)
|
|
- float(row.get("sell_lg_amount") or 0) - float(row.get("sell_elg_amount") or 0)
|
|
)
|
|
medium_net = float(row.get("buy_md_amount") or 0) - float(row.get("sell_md_amount") or 0)
|
|
small_net = float(row.get("buy_sm_amount") or 0) - float(row.get("sell_sm_amount") or 0)
|
|
values.append((
|
|
str(row["trade_date"]), row["ts_code"], float(row.get("net_mf_amount") or 0),
|
|
large_net, medium_net, small_net,
|
|
))
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO moneyflow_daily
|
|
(trade_date, ts_code, net_mf_amount, large_net_amount, medium_net_amount, small_net_amount)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
|
|
net_mf_amount=excluded.net_mf_amount, large_net_amount=excluded.large_net_amount,
|
|
medium_net_amount=excluded.medium_net_amount, small_net_amount=excluded.small_net_amount
|
|
""",
|
|
values,
|
|
)
|
|
return len(values)
|
|
|
|
def factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
|
|
where = "WHERE trade_date <= ?" if end_date else ""
|
|
parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
f"SELECT DISTINCT trade_date FROM daily_bars {where} ORDER BY trade_date DESC LIMIT ?",
|
|
parameters,
|
|
).fetchall()
|
|
return [row["trade_date"] for row in reversed(rows)]
|
|
|
|
def load_factor_data(self, end_date: str, limit_dates: int = 80) -> dict[str, Any]:
|
|
dates = self.factor_dates(end_date, limit_dates)
|
|
if not dates:
|
|
return {"dates": [], "bars": [], "master": [], "indicators": [], "moneyflow": []}
|
|
placeholders = ",".join("?" for _ in dates)
|
|
with self.connect() as connection:
|
|
bars = connection.execute(
|
|
f"SELECT * FROM daily_bars WHERE trade_date IN ({placeholders}) ORDER BY trade_date, ts_code",
|
|
dates,
|
|
).fetchall()
|
|
master = connection.execute("SELECT * FROM stock_master").fetchall()
|
|
indicators = connection.execute(
|
|
"""
|
|
SELECT * FROM daily_indicators
|
|
WHERE trade_date = (
|
|
SELECT MAX(trade_date) FROM daily_indicators WHERE trade_date <= ?
|
|
)
|
|
""",
|
|
(end_date,),
|
|
).fetchall()
|
|
moneyflow = connection.execute(
|
|
"""
|
|
SELECT * FROM moneyflow_daily
|
|
WHERE trade_date = (
|
|
SELECT MAX(trade_date) FROM moneyflow_daily WHERE trade_date <= ?
|
|
)
|
|
""",
|
|
(end_date,),
|
|
).fetchall()
|
|
return {
|
|
"dates": dates,
|
|
"bars": [dict(row) for row in bars],
|
|
"master": [dict(row) for row in master],
|
|
"indicators": [dict(row) for row in indicators],
|
|
"moneyflow": [dict(row) for row in moneyflow],
|
|
}
|
|
|
|
def snapshot_summaries(self, end_date: str, limit: int = 10) -> list[dict[str, Any]]:
|
|
try:
|
|
from sentiment_engine import build_sentiment_history
|
|
except ModuleNotFoundError:
|
|
from .sentiment_engine import build_sentiment_history
|
|
|
|
series = build_sentiment_history(self.list_snapshot_payloads(end_date, 240))
|
|
return [
|
|
{
|
|
"trade_date": row["trade_date"],
|
|
"sentiment_score": row["score"],
|
|
"seal_rate": row["seal_rate"],
|
|
"limit_up_count": row["limit_up_count"],
|
|
"limit_down_count": row["limit_down_count"],
|
|
"broken_count": row["broken_count"],
|
|
"up_count": row["up_count"],
|
|
"down_count": row["down_count"],
|
|
"amount_billion": row["amount_billion"],
|
|
}
|
|
for row in series[-limit:]
|
|
]
|
|
|
|
def list_snapshot_payloads(self, end_date: str, limit: int = 240) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT trade_date, payload FROM dashboard_snapshots
|
|
WHERE trade_date <= ? ORDER BY trade_date DESC LIMIT ?
|
|
""",
|
|
(end_date, limit),
|
|
).fetchall()
|
|
result: list[dict[str, Any]] = []
|
|
for row in reversed(rows):
|
|
try:
|
|
payload = json.loads(row["payload"])
|
|
except json.JSONDecodeError:
|
|
continue
|
|
payload["_snapshot_date"] = row["trade_date"]
|
|
result.append(payload)
|
|
return result
|
|
|
|
def save_screener_strategy(
|
|
self, user_id: int | None, name: str, description: str, regimes: list[str], formula: dict[str, Any],
|
|
builtin: bool = False, strategy_id: int | None = None,
|
|
) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
regimes_json = json.dumps(regimes, ensure_ascii=False)
|
|
formula_json = json.dumps(formula, ensure_ascii=False, separators=(",", ":"))
|
|
with self.connect() as connection:
|
|
if strategy_id:
|
|
if builtin:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE screener_strategies SET name=?, description=?, regimes=?, formula=?,
|
|
builtin=1, user_id=NULL, updated_at=? WHERE id=? AND builtin=1
|
|
""",
|
|
(name, description, regimes_json, formula_json, now, strategy_id),
|
|
)
|
|
else:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE screener_strategies SET name=?, description=?, regimes=?, formula=?,
|
|
updated_at=? WHERE id=? AND builtin=0 AND user_id=?
|
|
""",
|
|
(name, description, regimes_json, formula_json, now, strategy_id, int(user_id or 0)),
|
|
)
|
|
if cursor.rowcount == 0:
|
|
raise ValueError("选股策略不存在。")
|
|
return strategy_id
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO screener_strategies
|
|
(user_id, name, description, regimes, formula, builtin, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(None if builtin else int(user_id or 0), name, description, regimes_json, formula_json, int(builtin), now, now),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def list_screener_strategies(self, user_id: int | None = None) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
if user_id is None:
|
|
rows = connection.execute(
|
|
"SELECT * FROM screener_strategies WHERE builtin = 1 ORDER BY updated_at DESC, id"
|
|
).fetchall()
|
|
else:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT * FROM screener_strategies
|
|
WHERE builtin = 1 OR user_id = ?
|
|
ORDER BY builtin DESC, updated_at DESC, id
|
|
""",
|
|
(int(user_id),),
|
|
).fetchall()
|
|
result = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["regimes"] = json.loads(item["regimes"])
|
|
item["formula"] = json.loads(item["formula"])
|
|
item["builtin"] = bool(item["builtin"])
|
|
result.append(item)
|
|
return result
|
|
|
|
def delete_screener_strategy(self, user_id: int, strategy_id: int) -> bool:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT builtin, user_id FROM screener_strategies WHERE id = ?",
|
|
(strategy_id,),
|
|
).fetchone()
|
|
if not row:
|
|
raise ValueError("选股策略不存在。")
|
|
if bool(row["builtin"]):
|
|
raise ValueError("内置策略不能删除。")
|
|
if int(row["user_id"] or 0) != int(user_id):
|
|
raise ValueError("无权删除其他账号的策略。")
|
|
cursor = connection.execute(
|
|
"DELETE FROM screener_strategies WHERE id = ? AND builtin = 0 AND user_id = ?",
|
|
(strategy_id, int(user_id)),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def save_screener_run(
|
|
self, user_id: int, trade_date: str, regime: str, strategy_name: str,
|
|
formula: dict[str, Any], result: dict[str, Any],
|
|
) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO screener_runs
|
|
(user_id, trade_date, regime, strategy_name, formula, result, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
(int(user_id), trade_date, regime, strategy_name,
|
|
json.dumps(formula, ensure_ascii=False, separators=(",", ":")),
|
|
json.dumps(result, ensure_ascii=False, separators=(",", ":")), now),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def latest_screener_run(self, user_id: int, trade_date: str) -> dict[str, Any] | None:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT id, trade_date, regime, strategy_name, result, created_at
|
|
FROM screener_runs WHERE user_id = ? AND trade_date <= ? ORDER BY id DESC LIMIT 1
|
|
""",
|
|
(int(user_id), trade_date),
|
|
).fetchone()
|
|
if not row:
|
|
return None
|
|
try:
|
|
result = json.loads(row["result"])
|
|
except json.JSONDecodeError:
|
|
return None
|
|
result.setdefault("meta", {})["run_id"] = row["id"]
|
|
result["meta"]["created_at"] = row["created_at"]
|
|
return result
|
|
|
|
def save_mentor_exchange(
|
|
self,
|
|
user_id: int,
|
|
mentor_id: str,
|
|
trade_date: str,
|
|
question: str,
|
|
answer: str,
|
|
meta: str = "",
|
|
) -> None:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO mentor_messages
|
|
(user_id, mentor_id, trade_date, role, content, meta, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?)
|
|
""",
|
|
[
|
|
(int(user_id), mentor_id, trade_date, "user", question, "", now),
|
|
(int(user_id), mentor_id, trade_date, "assistant", answer, meta, now),
|
|
],
|
|
)
|
|
connection.execute(
|
|
"""
|
|
DELETE FROM mentor_messages
|
|
WHERE user_id = ? AND id NOT IN (
|
|
SELECT id FROM mentor_messages WHERE user_id = ? ORDER BY id DESC LIMIT 500
|
|
)
|
|
""",
|
|
(int(user_id), int(user_id)),
|
|
)
|
|
|
|
def list_mentor_messages(
|
|
self, user_id: int, mentor_id: str, trade_date: str, limit: int = 100
|
|
) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT role, content, meta, created_at FROM mentor_messages
|
|
WHERE user_id = ? AND mentor_id = ? AND trade_date = ?
|
|
ORDER BY id DESC LIMIT ?
|
|
""",
|
|
(int(user_id), mentor_id, trade_date, max(1, min(500, int(limit)))),
|
|
).fetchall()
|
|
return [dict(row) for row in reversed(rows)]
|
|
|
|
def delete_mentor_messages(self, user_id: int, mentor_id: str, trade_date: str) -> int:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM mentor_messages WHERE user_id = ? AND mentor_id = ? AND trade_date = ?",
|
|
(int(user_id), mentor_id, trade_date),
|
|
)
|
|
return int(cursor.rowcount)
|
|
|
|
def save_strategy_tracks(
|
|
self,
|
|
user_id: int,
|
|
run_id: int,
|
|
selection_date: str,
|
|
strategy_name: str,
|
|
candidates: list[dict[str, Any]],
|
|
) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
values = []
|
|
for item in candidates:
|
|
ts_code = str(item.get("ts_code") or "").strip()
|
|
code = str(item.get("code") or ts_code.split(".")[0]).strip()
|
|
entry_price = float(item.get("price") or 0)
|
|
if not ts_code or not code or entry_price <= 0:
|
|
continue
|
|
values.append(
|
|
(
|
|
int(user_id), int(run_id), selection_date, strategy_name, ts_code, code,
|
|
str(item.get("name") or "--"), str(item.get("sector") or "其他"),
|
|
entry_price, now, now,
|
|
)
|
|
)
|
|
with self.connect() as connection:
|
|
connection.executemany(
|
|
"""
|
|
INSERT INTO strategy_tracks
|
|
(user_id, run_id, selection_date, strategy_name, ts_code, code,
|
|
name, sector, entry_price, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(user_id, run_id, ts_code) DO UPDATE SET
|
|
name=excluded.name, sector=excluded.sector,
|
|
entry_price=excluded.entry_price, updated_at=excluded.updated_at
|
|
""",
|
|
values,
|
|
)
|
|
return len(values)
|
|
|
|
def list_strategy_tracks(self, user_id: int, limit_batches: int = 12) -> list[dict[str, Any]]:
|
|
limit_batches = max(1, min(50, int(limit_batches)))
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT * FROM strategy_tracks
|
|
WHERE user_id = ? AND run_id IN (
|
|
SELECT run_id FROM strategy_tracks WHERE user_id = ?
|
|
GROUP BY run_id ORDER BY run_id DESC LIMIT ?
|
|
)
|
|
ORDER BY run_id DESC, id
|
|
""",
|
|
(int(user_id), int(user_id), limit_batches),
|
|
).fetchall()
|
|
return [dict(row) for row in rows]
|
|
|
|
def load_tracking_bars(
|
|
self, targets: list[tuple[str, str]], limit: int = 5
|
|
) -> dict[tuple[str, str], list[dict[str, Any]]]:
|
|
unique_targets = set(targets)
|
|
if not unique_targets:
|
|
return {}
|
|
codes = sorted({ts_code for ts_code, _ in unique_targets})
|
|
earliest_date = min(selection_date for _, selection_date in unique_targets)
|
|
placeholders = ",".join("?" for _ in codes)
|
|
with self.connect() as connection:
|
|
rows = connection.execute(
|
|
f"""
|
|
SELECT ts_code, trade_date, open, high, low, close FROM daily_bars
|
|
WHERE ts_code IN ({placeholders}) AND trade_date > ?
|
|
ORDER BY ts_code, trade_date
|
|
""",
|
|
[*codes, earliest_date],
|
|
).fetchall()
|
|
by_code: dict[str, list[dict[str, Any]]] = {}
|
|
for row in rows:
|
|
item = dict(row)
|
|
by_code.setdefault(str(item["ts_code"]), []).append(item)
|
|
row_limit = max(1, min(20, int(limit)))
|
|
return {
|
|
(ts_code, selection_date): [
|
|
row for row in by_code.get(ts_code, []) if row["trade_date"] > selection_date
|
|
][:row_limit]
|
|
for ts_code, selection_date in unique_targets
|
|
}
|
|
|
|
def save_alert(
|
|
self,
|
|
user_id: int,
|
|
kind: str,
|
|
title: str,
|
|
content: str,
|
|
available_date: str,
|
|
code: str,
|
|
dedupe_key: str,
|
|
) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO alerts
|
|
(user_id, kind, title, content, available_date, code, dedupe_key,
|
|
is_read, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, 0, ?, ?)
|
|
ON CONFLICT(user_id, dedupe_key) DO UPDATE SET
|
|
title=excluded.title, content=excluded.content,
|
|
available_date=excluded.available_date, updated_at=excluded.updated_at
|
|
""",
|
|
(
|
|
int(user_id), kind, title, content, available_date, code,
|
|
dedupe_key, now, now,
|
|
),
|
|
)
|
|
row = connection.execute(
|
|
"SELECT id FROM alerts WHERE user_id = ? AND dedupe_key = ?",
|
|
(int(user_id), dedupe_key),
|
|
).fetchone()
|
|
return int(row["id"])
|
|
|
|
def list_alerts(
|
|
self, user_id: int, as_of: str, unread_only: bool = False, limit: int = 100
|
|
) -> list[dict[str, Any]]:
|
|
with self.connect() as connection:
|
|
if unread_only:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT id, kind, title, content, available_date, code, is_read,
|
|
created_at, updated_at, read_at
|
|
FROM alerts
|
|
WHERE user_id = ? AND available_date <= ? AND is_read = 0
|
|
ORDER BY available_date DESC, id DESC LIMIT ?
|
|
""",
|
|
(int(user_id), as_of, max(1, min(300, int(limit)))),
|
|
).fetchall()
|
|
else:
|
|
rows = connection.execute(
|
|
"""
|
|
SELECT id, kind, title, content, available_date, code, is_read,
|
|
created_at, updated_at, read_at
|
|
FROM alerts WHERE user_id = ?
|
|
ORDER BY CASE WHEN available_date > ? THEN 0 ELSE 1 END,
|
|
is_read, available_date, id DESC LIMIT ?
|
|
""",
|
|
(int(user_id), as_of, max(1, min(300, int(limit)))),
|
|
).fetchall()
|
|
return [{**dict(row), "is_read": bool(row["is_read"])} for row in rows]
|
|
|
|
def count_unread_alerts(self, user_id: int, as_of: str) -> int:
|
|
with self.connect() as connection:
|
|
row = connection.execute(
|
|
"""
|
|
SELECT COUNT(*) AS total FROM alerts
|
|
WHERE user_id = ? AND available_date <= ? AND is_read = 0
|
|
""",
|
|
(int(user_id), as_of),
|
|
).fetchone()
|
|
return int(row["total"] if row else 0)
|
|
|
|
def mark_alert_read(self, user_id: int, alert_id: int) -> bool:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE alerts SET is_read = 1, read_at = ?, updated_at = ?
|
|
WHERE id = ? AND user_id = ?
|
|
""",
|
|
(now, now, int(alert_id), int(user_id)),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def mark_all_alerts_read(self, user_id: int, as_of: str) -> int:
|
|
now = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
UPDATE alerts SET is_read = 1, read_at = ?, updated_at = ?
|
|
WHERE user_id = ? AND available_date <= ? AND is_read = 0
|
|
""",
|
|
(now, now, int(user_id), as_of),
|
|
)
|
|
return int(cursor.rowcount)
|
|
|
|
def delete_alert(self, user_id: int, alert_id: int) -> bool:
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"DELETE FROM alerts WHERE id = ? AND user_id = ?",
|
|
(int(alert_id), int(user_id)),
|
|
)
|
|
return cursor.rowcount > 0
|
|
|
|
def start_sync(self, trade_date: str, source: str) -> int:
|
|
started_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
cursor = connection.execute(
|
|
"""
|
|
INSERT INTO sync_runs (trade_date, source, status, started_at)
|
|
VALUES (?, ?, 'running', ?)
|
|
""",
|
|
(trade_date, source, started_at),
|
|
)
|
|
return int(cursor.lastrowid)
|
|
|
|
def finish_sync(
|
|
self,
|
|
sync_id: int,
|
|
status: str,
|
|
record_count: int = 0,
|
|
message: str = "",
|
|
source: str | None = None,
|
|
) -> None:
|
|
finished_at = datetime.now().astimezone().isoformat(timespec="seconds")
|
|
with self.connect() as connection:
|
|
connection.execute(
|
|
"""
|
|
UPDATE sync_runs
|
|
SET status = ?, finished_at = ?, record_count = ?, message = ?,
|
|
source = COALESCE(?, source)
|
|
WHERE id = ?
|
|
""",
|
|
(status, finished_at, record_count, message[:1000], source, sync_id),
|
|
)
|
|
|
|
def status(self) -> dict[str, Any]:
|
|
with self.connect() as connection:
|
|
last_sync = connection.execute(
|
|
"""
|
|
SELECT id, trade_date, source, status, started_at, finished_at,
|
|
record_count, message
|
|
FROM sync_runs ORDER BY id DESC LIMIT 1
|
|
"""
|
|
).fetchone()
|
|
snapshot_stats = connection.execute(
|
|
"""
|
|
SELECT COUNT(*) AS dates, COALESCE(SUM(record_count), 0) AS records,
|
|
MAX(updated_at) AS updated_at
|
|
FROM dashboard_snapshots
|
|
"""
|
|
).fetchone()
|
|
watchlist_count = connection.execute("SELECT COUNT(*) FROM watchlist").fetchone()[0]
|
|
note_count = connection.execute("SELECT COUNT(*) FROM review_notes").fetchone()[0]
|
|
|
|
return {
|
|
"database": str(self.path.name),
|
|
"snapshot_dates": int(snapshot_stats["dates"]),
|
|
"snapshot_records": int(snapshot_stats["records"]),
|
|
"updated_at": snapshot_stats["updated_at"],
|
|
"last_sync": dict(last_sync) if last_sync else None,
|
|
"watchlist_count": int(watchlist_count),
|
|
"note_count": int(note_count),
|
|
}
|