scanner/paths.py

"""Centralised filesystem paths.

Hugging Face Spaces only allow writing under ``/tmp``.  Locally (esp. on
Windows) ``/tmp`` does not exist, so we fall back to :func:`tempfile.gettempdir`.
All write-paths in the project should resolve through this module.
"""

from __future__ import annotations

import os
import tempfile


def _base_tmp() -> str:
    explicit = os.environ.get("FSCANNER_TMP_DIR", "").strip()
    if explicit:
        os.makedirs(explicit, exist_ok=True)
        return explicit
    # Prefer /tmp when it exists (HF Spaces); otherwise the OS temp dir.
    if os.path.isdir("/tmp"):
        return "/tmp"
    return tempfile.gettempdir()


TMP_DIR: str = _base_tmp()

CACHE_PATH: str = os.environ.get(
    "FSCANNER_CACHE_PATH",
    os.path.join(TMP_DIR, "us_flow_scanner_ohlcv.parquet"),
)

HISTORY_DIR: str = os.environ.get(
    "FSCANNER_HISTORY_DIR",
    os.path.join(TMP_DIR, "us_flow_scanner_history"),
)

SECTOR_CACHE_PATH: str = os.environ.get(
    "FSCANNER_SECTOR_CACHE",
    os.path.join(TMP_DIR, "us_flow_scanner_sectors.parquet"),
)

WATCHLIST_PATH: str = os.environ.get(
    "FSCANNER_WATCHLIST",
    os.path.join(TMP_DIR, "us_flow_scanner_watchlist.json"),
)

LEARNED_WEIGHTS_PATH: str = os.environ.get(
    "FSCANNER_LEARNED_WEIGHTS",
    os.path.join(TMP_DIR, "us_flow_scanner_learned_weights.json"),
)

PERFORMANCE_LOG_PATH: str = os.environ.get(
    "FSCANNER_PERFORMANCE_LOG",
    os.path.join(TMP_DIR, "us_flow_scanner_performance.parquet"),
)

RESULTS_CSV_PATH: str = os.path.join(TMP_DIR, "us_flow_scan_results.csv")

L2_CACHE_PATH: str = os.environ.get(
    "FSCANNER_L2_CACHE",
    os.path.join(TMP_DIR, "us_flow_scanner_l2.parquet"),
)

OPTIONS_CACHE_PATH: str = os.environ.get(
    "FSCANNER_OPTIONS_CACHE",
    os.path.join(TMP_DIR, "us_flow_scanner_options.parquet"),
)

TICK_CACHE_DIR: str = os.environ.get(
    "FSCANNER_TICK_CACHE",
    os.path.join(TMP_DIR, "us_flow_scanner_ticks"),
)

INTRADAY_CACHE_DIR: str = os.environ.get(
    "FSCANNER_INTRADAY_CACHE",
    os.path.join(TMP_DIR, "us_flow_scanner_intraday"),
)

STUB_DIR: str = os.environ.get(
    "FSCANNER_STUB_DIR",
    os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "stubs"),
)


def ensure_dirs() -> None:
    for d in (TMP_DIR, HISTORY_DIR, TICK_CACHE_DIR, INTRADAY_CACHE_DIR):
        try:
            os.makedirs(d, exist_ok=True)
        except OSError:
            pass


ensure_dirs()