Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,9 @@ The first stable open-source release.
- The data-sync config loads without a `consumer_config.ini`.
- Cache files with two-word categories are filed under the right ticker.
- `/api/bands_replay` reads the analysis cache instead of a fixed developer folder.
- Every module resolves the per-user folder the same way, so Pulse Room finds the cone and band
files the sync client and the local engine write (it looked in another folder outside the desktop
app). Files from the old locations are copied in once, never moved or overwritten.
- Pydantic v2 deprecation warnings are gone.

## [0.1.0-alpha] — 2026-05-21
Expand Down
25 changes: 10 additions & 15 deletions backend/account_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,13 +33,18 @@

import json
import logging
import os
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional
from datetime import datetime, date, timedelta
from pathlib import Path

# version: v4 (broker-TZ-aware period-pnl bucketing)
import app_paths

# version: v5 (user-data dir from app_paths)
# v5 — _get_user_data_dir() returns app_paths.app_data_dir(), shared with
# every other module. Off Windows that is ~/.<APP_DIR>/ (exact case)
# rather than ~/.<app_dir lowercased>/; app_paths copies existing
# account_settings.json / account_state.json over once.
# v4 — get_period_pnl now corrects for the broker TZ before bucketing
# deals into today/yesterday/this_week/etc. The MetaTrader5 library
# returns deal.time as broker server time treated as a Unix epoch
Expand All @@ -55,23 +60,13 @@

PROJECT_ROOT = Path(__file__).resolve().parent

# v3: AppData dir name from env var so v1 / v2 builds get separate dirs.
APP_DIR = os.environ.get("MK_APP_DIR_NAME", "QuantumTerminal")


def _get_user_data_dir() -> Path:
"""Per-user writable dir for account_settings.json + account_state.json.
Fixes Errno 13 Permission denied under Program Files on installed consumer.
v3: AppData dir name comes from MK_APP_DIR_NAME (default "QuantumTerminal")."""
import platform as _platform
if _platform.system() == "Windows":
appdata = os.environ.get("APPDATA", "")
base = Path(appdata) / APP_DIR if appdata else \
Path.home() / "AppData" / "Roaming" / APP_DIR
else:
base = Path.home() / f".{APP_DIR.lower()}"
base.mkdir(parents=True, exist_ok=True)
return base
v3: AppData dir name comes from MK_APP_DIR_NAME (default "QuantumTerminal").
v5: resolved by app_paths.app_data_dir()."""
return app_paths.app_data_dir()


def _resolve_default_settings_path() -> Path:
Expand Down
151 changes: 151 additions & 0 deletions backend/app_paths.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,151 @@
"""
================================================================================
Quantum Terminal — per-user app-data and cache directories
================================================================================
Single source of truth for where the backend keeps per-user files. Every module
that reads or writes them (sync client, cone/regime/stress-lab routes, Pulse
Room, config, account state, logs, cache watcher) resolves its folder here, so
they can never disagree about which directory holds the analysis files.

Windows: %APPDATA%\\<name>\\ (APPDATA unset: ~\\AppData\\Roaming\\<name>\\)
Linux/macOS: ~/.<name>/
cache: <app-data dir>/cache/

<name> is MK_APP_DIR_NAME when set, else "QuantumTerminal". The Electron shell
passes MK_APP_DIR_NAME=QuantumTerminal-v2, so the desktop app keeps using the
%APPDATA%\\QuantumTerminal-v2\\ data it already has, apart from browser-mode
(`python launcher.py`) installs, which keep %APPDATA%\\QuantumTerminal\\.

Paths are resolved on every call, not at import, so the process environment
(or a test's monkeypatched one) always wins.

Legacy locations: earlier builds kept a few files elsewhere. Pulse Room
defaulted to "QuantumTerminal-v2" (under ~/.config off Windows), and off
Windows the account files (including the simulator's paper account),
consumer_config.ini and user_config.json each had their own folder. The first time a process resolves the app-data dir, any of
those files missing from it are copied in — never moved, never overwritten —
and a marker file stops the copy from running again.
================================================================================
"""

import logging
import os
import shutil
import sys
import threading
from datetime import datetime, timezone
from pathlib import Path

log = logging.getLogger("qt.app_paths")

DEFAULT_APP_DIR_NAME = "QuantumTerminal"

# Pulse Room's own default before it resolved through this module.
_OLD_PULSE_DIR_NAME = "QuantumTerminal-v2"
_PULSE_STATE_FILES = ("pulse_watchlist.json", "pulse_state.json")
_MIGRATION_MARKER = ".migrated_legacy_paths"

_migration_lock = threading.Lock()
_migrated_targets: set = set()


def _is_windows() -> bool:
return sys.platform == "win32"


def app_dir_name() -> str:
"""MK_APP_DIR_NAME if set and not blank, else "QuantumTerminal"."""
return os.environ.get("MK_APP_DIR_NAME", "").strip() or DEFAULT_APP_DIR_NAME


def app_data_dir_for(name: str) -> Path:
"""Where the app-data dir called `name` lives on this platform. Creates
nothing — use app_data_dir() for this process's own directory."""
if _is_windows():
appdata = os.environ.get("APPDATA", "")
base = Path(appdata) if appdata else Path.home() / "AppData" / "Roaming"
return base / name
return Path.home() / f".{name}"


def app_data_dir() -> Path:
"""This process's per-user app-data dir, created if missing."""
name = app_dir_name()
target = app_data_dir_for(name)
target.mkdir(parents=True, exist_ok=True)
if target not in _migrated_targets:
_copy_legacy_files(target, name)
return target


def cache_dir() -> Path:
"""The cache dir every reader and writer shares: <app-data dir>/cache."""
d = app_data_dir() / "cache"
d.mkdir(parents=True, exist_ok=True)
return d


# ── Legacy locations ─────────────────────────────────────────────────────────

def _legacy_locations(name: str) -> list:
"""(directory, filenames) pairs where earlier builds kept files that now
belong in app_data_dir_for(name)."""
# Pulse Room honoured MK_APP_DIR_NAME but had its own default name.
if os.environ.get("MK_APP_DIR_NAME", "").strip():
pulse_name = name
else:
pulse_name = _OLD_PULSE_DIR_NAME
if _is_windows():
# Every other module already used %APPDATA%\<name>.
return [(app_data_dir_for(pulse_name), _PULSE_STATE_FILES)]
home = Path.home()
return [
# pulse_room: ~/.config/<name>
(home / ".config" / pulse_name, _PULSE_STATE_FILES),
# account_manager, the simulator's paper account, and launcher's
# consumer_config.ini fallback
(home / f".{name.lower()}",
("account_settings.json", "account_state.json", "sim_paper_account.json",
"consumer_config.ini")),
# config_manager used the Windows layout under $HOME
(home / "AppData" / "Roaming" / name, ("user_config.json",)),
]


def _copy_legacy_files(target: Path, name: str) -> None:
"""Copy files missing from `target` in from their legacy locations. Never
moves or overwrites anything; failures are logged and retried next run."""
with _migration_lock:
if target in _migrated_targets:
return
_migrated_targets.add(target)

marker = target / _MIGRATION_MARKER
if marker.exists():
return
sources = [(d, files) for d, files in _legacy_locations(name)
if d != target and d.is_dir()]
if not sources:
return

copied, failed = [], False
for src_dir, files in sources:
for fname in files:
src, dst = src_dir / fname, target / fname
try:
if src.is_file() and not dst.exists():
shutil.copy2(src, dst)
copied.append(f"{src} -> {dst}")
log.info(f"Copied legacy {src} -> {dst}")
except OSError as e:
failed = True
log.warning(f"Could not copy legacy {src} -> {dst}: {e}")
if failed:
return # no marker, so the next start retries what's still missing

stamp = datetime.now(timezone.utc).isoformat(timespec="seconds")
try:
marker.write_text("\n".join([f"migrated {stamp}", *copied]) + "\n",
encoding="utf-8")
except OSError as e:
log.warning(f"Could not write {marker}: {e}")
26 changes: 11 additions & 15 deletions backend/cache_watcher.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
# version: v3
# version: v4
"""
================================================================================
Quantum Terminal Consumer — Cache Directory Watcher
Expand Down Expand Up @@ -51,6 +51,10 @@
correct in-memory slot. Same resolver used by _load_all_from_cache at
startup, ensuring watcher and loader stay consistent.

v4 — Watched dir and the consumer_config.ini fallback both come from
app_paths (shared with data_sync_client and Pulse Room), on every
platform rather than only when APPDATA is set.

Rule C1 compliance: this module performs ZERO calculations — it only reads
files already on disk and pushes WS events.
================================================================================
Expand All @@ -60,15 +64,13 @@
import configparser
import json
import logging
import os

# v2: AppData dir name (env-var driven; default "QuantumTerminal" preserves v1 path).
APP_DIR = os.environ.get("MK_APP_DIR_NAME", "QuantumTerminal")
import re
from datetime import datetime, timezone
from pathlib import Path
from typing import Awaitable, Callable, Optional

import app_paths # v4: per-user app-data + cache dirs, shared by every module

log = logging.getLogger("qt.cache_watcher")

DEFAULT_INTERVAL_SECONDS = 1.0
Expand All @@ -93,9 +95,7 @@ def _read_config() -> tuple[bool, float]:
if getattr(_sys, "frozen", False):
candidates.append(Path(_sys.executable).parent / "consumer_config.ini")

appdata = os.environ.get("APPDATA", "")
if appdata:
candidates.append(Path(appdata) / APP_DIR / "consumer_config.ini")
candidates.append(app_paths.app_data_dir() / "consumer_config.ini")

enabled = True
interval = DEFAULT_INTERVAL_SECONDS
Expand Down Expand Up @@ -123,15 +123,11 @@ def _read_config() -> tuple[bool, float]:


def _get_cache_dir() -> Optional[Path]:
"""Resolve %APPDATA%\\QuantumTerminal\\cache via the same helper data_sync_client uses."""
"""Resolve the cache dir data_sync_client writes to (app_paths.cache_dir())."""
try:
from data_sync_client import _get_cache_dir as _dsc_cache_dir
return _dsc_cache_dir()
return app_paths.cache_dir()
except Exception:
appdata = os.environ.get("APPDATA", "")
if not appdata:
return None
return Path(appdata) / APP_DIR / "cache"
return None


def create_cache_watcher_task(
Expand Down
55 changes: 27 additions & 28 deletions backend/config_manager.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,11 @@
# version: v11
# version: v12
# v12 — AppData dir resolved by app_paths.app_data_dir(), shared by every
# module (2026-10-17). Off Windows, user_config.json moves from
# ~/AppData/Roaming/<APP_DIR>/ to ~/.<APP_DIR>/ (app_paths copies the
# old file over once), and _legacy_user_config_paths() now includes the
# live file on every platform — previously, off Windows, the v11
# migration found no sources and stripped the profiles unmigrated.
#
# v11 — User-owned saves move to %USERPROFILE%\Documents\Quantum Terminal\ as
# per-file stores (2026-05-13). Bug: a 4-key trimmed user_config.json
# was being written when _load_user_config silently fell back to {}
Expand Down Expand Up @@ -111,28 +118,23 @@
from typing import Dict, List, Optional, Any
from datetime import datetime, timezone

import app_paths

log = logging.getLogger("config_manager")

# ── Project root (Rule 3) ──
PROJECT_ROOT = Path(__file__).resolve().parent

# v6: AppData dir — same env-var convention used by license_client.py so v1
# and v2 can run side by side without sharing user state. Default
# "QuantumTerminal" preserves v1 behavior.
APP_DIR = os.environ.get("MK_APP_DIR_NAME", "QuantumTerminal")


# v6: AppData dir — MK_APP_DIR_NAME lets v1 and v2 run side by side without
# sharing user state; default "QuantumTerminal" preserves v1 behavior.
# v12: resolved by app_paths, like every other module.
def _appdata_user_config_path() -> Path:
"""Resolve %APPDATA%\\<APP_DIR>\\user_config.json. Used as the canonical
storage path for user preferences + saved profiles. Survives reinstalls
of the app (NSIS overwrites the install dir but leaves AppData alone)."""
appdata = os.environ.get("APPDATA")
if appdata:
base = Path(appdata) / APP_DIR
else:
base = Path.home() / "AppData" / "Roaming" / APP_DIR
base.mkdir(parents=True, exist_ok=True)
return base / "user_config.json"
"""Resolve <app-data dir>\\user_config.json (%APPDATA%\\<APP_DIR>\\ on
Windows; see app_paths). Used as the canonical storage path for user
preferences + saved profiles. Survives reinstalls of the app (NSIS
overwrites the install dir but leaves AppData alone)."""
return app_paths.app_data_dir() / "user_config.json"


# ════════════════════════════════════════════════════════════════
Expand Down Expand Up @@ -332,18 +334,15 @@ def _migration_marker_path() -> Path:
def _legacy_user_config_paths() -> List[Path]:
"""Return up to three paths to legacy user_config.json files, in
priority order: live AppData v2 first, then any sibling dirs."""
appdata = os.environ.get("APPDATA")
paths: List[Path] = []
if appdata:
base = Path(appdata)
# Live v2 first (most authoritative).
paths.append(base / APP_DIR / "user_config.json")
# QuantumTerminal-X — orphaned attempt with a typo'd env var.
if APP_DIR != "QuantumTerminal-X":
paths.append(base / "QuantumTerminal-X" / "user_config.json")
# v1 — read-only fallback (Rule 8: never write).
if APP_DIR != "QuantumTerminal":
paths.append(base / "QuantumTerminal" / "user_config.json")
app_dir = app_paths.app_dir_name()
# Live v2 first (most authoritative).
paths: List[Path] = [_appdata_user_config_path()]
# QuantumTerminal-X — orphaned attempt with a typo'd env var.
if app_dir != "QuantumTerminal-X":
paths.append(app_paths.app_data_dir_for("QuantumTerminal-X") / "user_config.json")
# v1 — read-only fallback (Rule 8: never write).
if app_dir != "QuantumTerminal":
paths.append(app_paths.app_data_dir_for("QuantumTerminal") / "user_config.json")
return [p for p in paths if p.exists()]


Expand Down
Loading
Loading