360 lines
12 KiB
Python
360 lines
12 KiB
Python
"""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():
|
|
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"])
|
|
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)
|