IB-Dashboard/backend/app.py

"""Flask backend for IB dashboard.

Provides simple REST endpoints that the vanilla-JS frontend polls:
- GET /api/quotes -> latest market quotes
- GET /api/positions -> latest positions
- POST /api/subscribe { symbols: ["AAPL","MSFT"] } -> start subscribing

This app expects an IB Gateway or TWS running locally on port 7496.
"""
from flask import Flask, jsonify, request, send_from_directory
import os
import threading

from flex_lots import load_latest_flex_lots
from flex_query import FlexQueryError, fetch_flex_statement, get_flex_status
from ib_client import IBClient
import time

FLEX_AUTO_SYNC_MAX_AGE_SECONDS = 8 * 60 * 60
_flex_sync_lock = threading.Lock()
_flex_startup_sync_started = False
_flex_sync_state = {
    "last_attempt_at": None,
    "last_success_at": None,
    "last_result": None,
    "last_error": None,
    "last_saved_to": None,
    "last_reference_code": None,
    "last_trigger": None,
}


def load_dotenv_file() -> None:
    env_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".env"))
    if not os.path.exists(env_path):
        return
    try:
        with open(env_path, "r", encoding="utf-8") as handle:
            for raw_line in handle:
                line = raw_line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                key, value = line.split("=", 1)
                key = key.strip()
                value = value.strip().strip('"').strip("'")
                if key and key not in os.environ:
                    os.environ[key] = value
    except Exception:
        pass


load_dotenv_file()


# Use an absolute path for the static folder so Flask can reliably serve files
STATIC_FOLDER = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "frontend"))
app = Flask(__name__, static_folder=STATIC_FOLDER, static_url_path="/static")


def should_auto_sync_flex(status: dict) -> bool:
    if not status.get("configured"):
        return False
    latest_timestamp = status.get("latest_timestamp")
    if latest_timestamp is None:
        return True
    return (time.time() - latest_timestamp) > FLEX_AUTO_SYNC_MAX_AGE_SECONDS


def get_flex_status_with_sync_state() -> dict:
    status = get_flex_status()
    status.update(_flex_sync_state)
    return status


def run_flex_sync_if_needed(reason: str) -> bool:
    status = get_flex_status()
    if not should_auto_sync_flex(status):
        return False

    if not _flex_sync_lock.acquire(blocking=False):
        return False

    try:
        _flex_sync_state["last_attempt_at"] = time.time()
        _flex_sync_state["last_trigger"] = reason
        _flex_sync_state["last_result"] = "running"
        _flex_sync_state["last_error"] = None

        # Re-check after taking the lock so concurrent callers do not duplicate work.
        status = get_flex_status()
        if not should_auto_sync_flex(status):
            _flex_sync_state["last_result"] = "skipped"
            return False
        result = fetch_flex_statement()
        _flex_sync_state["last_success_at"] = time.time()
        _flex_sync_state["last_result"] = "success"
        _flex_sync_state["last_saved_to"] = result["saved_to"]
        _flex_sync_state["last_reference_code"] = result["reference_code"]
        print(
            f"Flex sync completed ({reason}): saved_to={result['saved_to']} "
            f"reference_code={result['reference_code']}"
        )
        return True
    except FlexQueryError as exc:
        _flex_sync_state["last_result"] = "error"
        _flex_sync_state["last_error"] = str(exc)
        print(f"Flex sync skipped/failed ({reason}): {exc}")
        return False
    except Exception as exc:
        _flex_sync_state["last_result"] = "error"
        _flex_sync_state["last_error"] = str(exc)
        print(f"Unexpected Flex sync failure ({reason}): {exc}")
        return False
    finally:
        _flex_sync_lock.release()


def start_flex_startup_sync_if_needed() -> None:
    global _flex_startup_sync_started

    if _flex_startup_sync_started:
        return
    _flex_startup_sync_started = True

    thread = threading.Thread(
        target=run_flex_sync_if_needed,
        args=("startup",),
        daemon=True,
    )
    thread.start()

# instantiate the IB client with a process-unique client id to avoid
# collisions when Flask's reloader spawns multiple processes.
ib = IBClient(client_id=os.getpid())
# Start connection only in the main process (Werkzeug sets WERKZEUG_RUN_MAIN)
if os.environ.get("WERKZEUG_RUN_MAIN") == "true" or not os.environ.get("WERKZEUG_RUN_MAIN"):
    # attempt to start the connection; failures are logged by the client
    try:
        ib.connect_and_start()
    except Exception:
        pass
    start_flex_startup_sync_if_needed()


@app.route("/")
def index():
    return send_from_directory(app.static_folder, "index.html")


@app.route("/api/quotes")
def api_quotes():
    return jsonify(ib.get_quotes())


@app.route("/api/positions")
def api_positions():
    # request a fresh positions snapshot from IB and wait briefly for callbacks
    try:
        ib.request_positions()
    except Exception:
        pass
    # wait up to 2 seconds for positions to arrive
    deadline = time.time() + 2.0
    while time.time() < deadline:
        positions = ib.get_positions()
        if positions:
            return jsonify(positions)
        time.sleep(0.1)
    # return whatever we have (possibly empty)
    return jsonify(ib.get_positions())


@app.route("/api/lots")
def api_lots():
    api_flex_sync()
    flex_status = get_flex_status()
    if flex_status.get("configured") and flex_status.get("data_dir"):
        latest_file, flex_lots = load_latest_flex_lots(flex_status["data_dir"])
        print(
            f"[D]Latest file{latest_file} "
        )

        if latest_file and flex_lots:
            return jsonify(flex_lots)

    # request a fresh positions snapshot first so all account holdings are covered,
    # even when IB does not return historical executions for every open position.
    try:
        ib.request_positions()
    except Exception:
        pass
    positions_deadline = time.time() + 2.0
    while time.time() < positions_deadline:
        if ib.get_positions():
            break
        time.sleep(0.1)

    for p in ib.get_positions():
        account = p.get("account") if isinstance(p, dict) else None
        if account:
            try:
                ib.request_account_updates(account)
            except Exception:
                pass

    # subscribe market data for all current positions to improve gain calculations
    for p in ib.get_positions():
        instrument_key = p.get("instrument_key") if isinstance(p, dict) else None
        if instrument_key:
            try:
                ib.subscribe_market_data(p)
            except Exception:
                pass
    quotes_deadline = time.time() + 2.0
    position_symbols = {
        p.get("instrument_key")
        for p in ib.get_positions()
        if isinstance(p, dict) and p.get("instrument_key")
    }
    while time.time() < quotes_deadline:
        quotes = ib.get_quotes()
        if any(instrument_key in quotes for instrument_key in position_symbols):
            break
        time.sleep(0.1)

    # request prior close for positions still missing a usable quote
    for p in ib.get_positions():
        instrument_key = p.get("instrument_key") if isinstance(p, dict) else None
        if not instrument_key:
            continue
        quote = ib.get_quotes().get(instrument_key, {})
        if quote.get("last") is None and quote.get("close") is None and quote.get("mark") is None:
            try:
                ib.request_historical_close(p)
            except Exception:
                pass
    historical_deadline = time.time() + 2.0
    while time.time() < historical_deadline:
        missing = []
        quotes = ib.get_quotes()
        for p in ib.get_positions():
            instrument_key = p.get("instrument_key") if isinstance(p, dict) else None
            if not instrument_key:
                continue
            quote = quotes.get(instrument_key, {})
            if (
                quote.get("last") is None
                and quote.get("close") is None
                and quote.get("mark") is None
                and instrument_key not in ib.historical_closes
            ):
                missing.append(instrument_key)
        if not missing:
            break
        time.sleep(0.1)

    # request executions and wait briefly for callbacks
    try:
        ib.request_executions(start_time="20000101 00:00:00")
    except Exception:
        pass
    deadline = time.time() + 3.0
    while time.time() < deadline:
        lots = ib.compute_open_lots(include_position_fallback=False)
        if lots:
            return jsonify(lots)
        time.sleep(0.1)
    return jsonify(ib.compute_open_lots(include_position_fallback=False))


@app.route("/api/executions-debug")
def api_executions_debug():
    start_time = request.args.get("start_time", "20000101 00:00:00")
    try:
        ib.request_executions(start_time=start_time)
    except Exception:
        pass
    deadline = time.time() + 5.0
    while time.time() < deadline:
        time.sleep(0.1)
    executions = list(ib.executions)
    summary = {}
    for e in executions:
        key = e.get("instrument_key") or e.get("symbol") or "UNKNOWN"
        item = summary.setdefault(key, {
            "instrument_key": e.get("instrument_key"),
            "symbol": e.get("symbol"),
            "security_type": e.get("security_type"),
            "display_symbol": e.get("display_symbol"),
            "count": 0,
            "first_time": None,
            "last_time": None,
        })
        item["count"] += 1
        etime = e.get("time")
        if etime:
            if item["first_time"] is None or etime < item["first_time"]:
                item["first_time"] = etime
            if item["last_time"] is None or etime > item["last_time"]:
                item["last_time"] = etime
    return jsonify({
        "start_time": start_time,
        "execution_count": len(executions),
        "executions": executions,
        "summary": list(summary.values()),
    })


@app.route("/api/flex/status")
def api_flex_status():
    return jsonify(get_flex_status_with_sync_state())


@app.route("/api/flex/lots-debug")
def api_flex_lots_debug():
    status = get_flex_status()
    if not status.get("configured") or not status.get("data_dir"):
        return jsonify({"configured": False, "lots": [], "latest_file": None})
    latest_file, lots = load_latest_flex_lots(status["data_dir"])
    return jsonify({
        "configured": True,
        "latest_file": latest_file,
        "lot_count": len(lots),
        "lots": lots,
    })


@app.route("/api/flex/sync", methods=["POST"])
def api_flex_sync():
    try:
        _flex_sync_state["last_attempt_at"] = time.time()
        _flex_sync_state["last_trigger"] = "manual"
        _flex_sync_state["last_result"] = "running"
        _flex_sync_state["last_error"] = None
        result = fetch_flex_statement()
        _flex_sync_state["last_success_at"] = time.time()
        _flex_sync_state["last_result"] = "success"
        _flex_sync_state["last_saved_to"] = result["saved_to"]
        _flex_sync_state["last_reference_code"] = result["reference_code"]
        return jsonify({"ok": True, **result})
    except FlexQueryError as exc:
        _flex_sync_state["last_result"] = "error"
        _flex_sync_state["last_error"] = str(exc)
        return jsonify({"ok": False, "error": str(exc)}), 400
    except Exception as exc:
        _flex_sync_state["last_result"] = "error"
        _flex_sync_state["last_error"] = str(exc)
        return jsonify({"ok": False, "error": str(exc)}), 500


@app.route("/api/subscribe", methods=["POST"])
def api_subscribe():
    data = request.get_json(force=True)
    symbols = data.get("symbols", []) if isinstance(data, dict) else []
    if not isinstance(symbols, list):
        return jsonify({"error": "symbols must be a list"}), 400
    for s in symbols:
        ib.subscribe_market_data(s)
    return jsonify({"subscribed": symbols})


if __name__ == "__main__":
    # simple dev server; disable reloader to avoid duplicate IB client instances
    app.run(debug=True, port=8000, use_reloader=False)