7d702e9cbd
- Add db.py PostgreSQL abstraction layer (connection, upsert, table mgmt) - Replace sqlite3 with psycopg2 in: live_candle_fetcher, resampler, data_fetcher, fetch_history, import_csv, indicators, base_strategy - Sanitize table names (colons -> underscores) for PostgreSQL compat - Replace INSERT OR REPLACE with ON CONFLICT upserts - Replace pandas to_sql() with batch upsert_candles() - Add scripts: resampler_loop, gap_detector, backup_runner, cron_scheduler - Add migrate_sqlite_to_pg.py for one-time data migration - Add Dockerfile, docker-compose.yml, supervisord.conf - Add postgres/postgresql.conf tuned for 4GB RAM (Synology DS1513+) - Add .dockerignore, .env.docker.example, secrets template - Update requirements.txt (psycopg2-binary), .gitignore - Add MIGRATION_PLAN.md with full plan and todo list
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
"""
|
|
Indicators Data Fetcher
|
|
|
|
A standalone process that runs in a loop to compute financial indicators
|
|
(ratios, prices, MAs, RSI, custom) from SQLite candle data and save
|
|
the results to a JSON status file for the main dashboard to display.
|
|
|
|
Follows the same pattern as dashboard_data_fetcher.py.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
import sys
|
|
import json
|
|
import time
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
|
|
from logging_utils import setup_logging
|
|
from indicators import IndicatorCalculator
|
|
|
|
|
|
class IndicatorsFetcher:
|
|
"""
|
|
Periodically computes all configured indicators and saves them to a JSON file.
|
|
"""
|
|
|
|
def __init__(self, log_level: str):
|
|
setup_logging(log_level, 'IndicatorsFetcher')
|
|
|
|
project_root = os.path.dirname(os.path.abspath(__file__))
|
|
self.db_path = os.environ.get("PG_CONN_STR", "postgresql://hyper:hyper@localhost:5432/hyper")
|
|
self.config_path = os.path.join(project_root, "_data", "indicators.json")
|
|
self.status_file_path = os.path.join(project_root, "_logs", "indicators_status.json")
|
|
|
|
self.calculator = IndicatorCalculator(
|
|
config_path=self.config_path,
|
|
db_path=self.db_path
|
|
)
|
|
|
|
logging.info(f"Indicators Fetcher initialized. DB: {self.db_path}, Config: {self.config_path}")
|
|
|
|
def fetch_and_save_indicators(self):
|
|
"""Compute all indicators and save to JSON status file."""
|
|
try:
|
|
results = self.calculator.calculate_all()
|
|
|
|
status = {
|
|
"last_updated_utc": datetime.now(timezone.utc).isoformat(),
|
|
"indicators": results
|
|
}
|
|
|
|
logs_dir = os.path.dirname(self.status_file_path)
|
|
os.makedirs(logs_dir, exist_ok=True)
|
|
|
|
temp_file_path = self.status_file_path + ".tmp"
|
|
with open(temp_file_path, 'w', encoding='utf-8') as f:
|
|
json.dump(status, f, indent=4, default=str)
|
|
os.replace(temp_file_path, self.status_file_path)
|
|
|
|
logging.debug(f"Successfully updated indicators status file with {len(results)} indicators.")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Failed to fetch or save indicators: {e}", exc_info=True)
|
|
|
|
def run(self):
|
|
"""Main loop to periodically compute and save indicators."""
|
|
logging.info("Starting Indicators Fetcher loop (update interval: 30s)")
|
|
while True:
|
|
try:
|
|
self.fetch_and_save_indicators()
|
|
except Exception as e:
|
|
logging.error(f"Indicators Fetcher loop error: {e}", exc_info=True)
|
|
time.sleep(30)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser(description="Run the Indicators Data Fetcher.")
|
|
parser.add_argument("--log-level", default="normal", choices=['off', 'normal', 'debug'])
|
|
args = parser.parse_args()
|
|
|
|
fetcher = IndicatorsFetcher(log_level=args.log_level)
|
|
try:
|
|
fetcher.run()
|
|
except KeyboardInterrupt:
|
|
logging.info("Indicators Data Fetcher stopped.")
|