File size: 2,437 Bytes
4d68493 6b66ac0 4d68493 6b66ac0 4d68493 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 | """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()
|