5 free tools for quant and options trading, wired into a strategy

from datetime import date, timedelta

import numpy as np
from openbb import obb

SYMBOL = "SPY"
since = (date.today() - timedelta(days=90)).isoformat()

# 1. option chain: every listed expiry and strike, with implied vol and Greeks
chain = obb.cboe.options.chains(SYMBOL).to_df()
spot = float(chain["underlying_price"].iloc[0])
expiry = chain.loc[(chain["dte"] - 30).abs().idxmin(), "expiration"]          # expiry closest to 30 days
month = chain[chain["expiration"] == expiry]
atm = month.loc[(month["strike"] - spot).abs().idxmin(), "strike"]
iv30 = month.loc[month["strike"] == atm, "implied_volatility"].mean()          # call and put at the money

# 2. realized vol over 21 trading days, annualized
close = obb.yfinance.equity.price.historical(SYMBOL, start_date=since).to_df()["close"]
rv21 = np.log(close).diff().tail(21).std() * np.sqrt(252)

# 3. macro: VIX and the 10 year Treasury yield
vix = obb.yfinance.index.price.historical("^VIX", start_date=since).to_df()["close"].iloc[-1]
ust10 = obb.federal_reserve.treasury_rates(start_date=since).to_df()["year_10"].iloc[-1]

# 4. fundamentals, for a single stock strategy
nvda = obb.yfinance.equity.fundamental.metrics("NVDA").to_df().iloc[0]

print(f"{SYMBOL} {spot:.2f}, {len(chain):,} contracts, expiry {expiry} "
      f"({int(month['dte'].iloc[0])} days), ATM strike {atm:g}")
print(f"implied vol 30d {iv30:.1%}, realized vol 21d {rv21:.1%}, ratio {iv30 / rv21:.2f}")
print(f"VIX {vix:.2f}, 10 year Treasury {ust10:.2%}")
print(f"NVDA forward P/E {nvda['forward_pe']:.1f}, operating margin {nvda['operating_margin']:.1%}")

# signal input: options rich against realized vol favour selling premium, cheap favour buying it
# (1.2 and 0.8 are placeholders, set them from your own backtest)
ratio = iv30 / rv21
print("IV/RV signal:", "sell premium" if ratio > 1.2 else "buy premium" if ratio < 0.8 else "no edge")

import asyncio
import json

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

SERVER = StdioServerParameters(command="uvx", args=["--from", "tradingview-mcp-server", "tradingview-mcp"])
WATCHLIST = ["AAPL", "MSFT", "NVDA", "AMZN", "META", "TSLA"]


async def call(tv, tool, **args):
    res = await tv.call_tool(tool, args)
    return [json.loads(c.text) for c in res.content]          # one JSON block per row


async def main():
    async with stdio_client(SERVER) as (read, write), ClientSession(read, write) as tv:
        await tv.initialize()
        print(len((await tv.list_tools()).tools), "tools on the server")
        keep = []
        for sym in WATCHLIST:
            await asyncio.sleep(2)                                    # TradingView throttles bursts of calls
            a = (await call(tv, "coin_analysis", symbol=sym, exchange="NASDAQ", timeframe="1D"))[0]
            if "error" in a:                                          # TradingView throttles, skip and move on
                err = a["error"]
                print(f"{sym:<5} skipped, {err.get('code', '') if isinstance(err, dict) else err}")
                continue
            close, sma200, rsi = a["price_data"]["close"], a["sma"]["sma200"], a["rsi"]["value"]
            bb = a["market_sentiment"]["overall_rating"]           # Bollinger position, -3 to +3
            ok = close > sma200 and rsi < 70 and bb >= 0              # uptrend, not overbought, upper half
            print(f"{sym:<5} close {close:>8.2f}  SMA200 {sma200:>8.2f}  RSI {rsi:5.1f}  BB {bb:+d}  "
                  f"{'keep' if ok else 'skip'}")
            keep.append(sym) if ok else None
        print("watchlist for the strategy:", keep)


asyncio.run(main())

methods = [   # (name, engine, which exercise styles it prices)
    ("Black-Scholes", ql.AnalyticEuropeanEngine(bsm), E),
    # Black Vasicek, method 2 of the C++ file, has no Python binding in QuantLib 1.44, so 20 of 21 here
    ("Heston semi-analytic", ql.AnalyticHestonEngine(heston), E),
    ("Bates semi-analytic", ql.BatesEngine(bates), E),
    ("Barone-Adesi/Whaley", ql.BaroneAdesiWhaleyApproximationEngine(bsm), A),
    ("Bjerksund/Stensland", ql.BjerksundStenslandApproximationEngine(bsm), A),
    ("QD+ fixed-point (fast)", qdfp(bsm, qdfp.fastScheme()), A),
    ("QD+ fixed-point (accurate)", qdfp(bsm, qdfp.accurateScheme()), A),
    ("QD+ fixed-point (high precision)", qdfp(bsm, qdfp.highPrecisionScheme()), A),
    ("Integral", ql.IntegralEngine(bsm), E),
    ("Finite differences", ql.FdBlackScholesVanillaEngine(bsm, steps, steps - 1), ALL),
] + [(name, ql.BinomialVanillaEngine(bsm, tree, steps), ALL) for name, tree in [

def quote_to_greeks(price, spot, strike, days, rate, div=0.0, right=ql.Option.Put):
    """Implied vol from a quoted European price, then delta, gamma, vega at that vol."""
    expiry = today + ql.Period(days, ql.Days)
    vol_q = ql.SimpleQuote(0.20)
    proc = ql.BlackScholesMertonProcess(
        ql.QuoteHandle(ql.SimpleQuote(spot)),
        ql.YieldTermStructureHandle(ql.FlatForward(today, div, dc)),
        ql.YieldTermStructureHandle(ql.FlatForward(today, rate, dc)),
        ql.BlackVolTermStructureHandle(ql.BlackConstantVol(today, cal, ql.QuoteHandle(vol_q), dc)))
    opt = ql.VanillaOption(ql.PlainVanillaPayoff(right, strike), ql.EuropeanExercise(expiry))
    opt.setPricingEngine(ql.AnalyticEuropeanEngine(proc))
    vol_q.setValue(opt.impliedVolatility(price, proc))
    return {"iv": vol_q.value(), "delta": opt.delta(), "gamma": opt.gamma(), "vega": opt.vega() / 100}


print("\nquote 4.00 on the same put:", {k: round(v, 4) for k, v in quote_to_greeks(4.00, 36, 40, 365, 0.06).items()})

import os
import sys
from datetime import date

import yfinance as yf
from tradingagents.default_config import DEFAULT_CONFIG
from tradingagents.graph.trading_graph import TradingAgentsGraph

# Buy, Overweight, Hold, Underweight, Sell; REVIEW (no readable rating) is not in the map, so no trade
SIDE = {"Buy": "BUY", "Overweight": "BUY", "Hold": None, "Underweight": "SELL", "Sell": "SELL"}


def tradingagents_signal(symbol, trade_date=None, provider=os.environ.get("TA_PROVIDER", "ollama")):
    config = DEFAULT_CONFIG.copy()
    config["max_debate_rounds"] = 1
    if provider == "ollama":                              # local model, free, through Ollama on port 11434
        config["llm_provider"] = "ollama"
        config["backend_url"] = "http://localhost:11434/v1"
        config["deep_think_llm"] = config["quick_think_llm"] = "qwen3:8b"
    else:                                                 # DeepSeek, about 6 cents a full run
        config["llm_provider"] = "deepseek"
        config["deep_think_llm"], config["quick_think_llm"] = "deepseek-v4-pro", "deepseek-v4-flash"
    ta = TradingAgentsGraph(debug=False, config=config)
    _, rating = ta.propagate(symbol, trade_date or date.today().isoformat())
    ref = float(yf.Ticker(symbol).history(period="5d")["Close"].iloc[-1])    # price the committee saw
    return SIDE.get(rating), ref


if __name__ == "__main__":
    sym = sys.argv[1] if len(sys.argv) > 1 else "NVDA"
    print(sym, tradingagents_signal(sym))                # ('BUY', last close), or (None, last close) on Hold

px = yf.download("SPY", start="2010-01-01", auto_adjust=True, progress=False)["Close"]["SPY"]
ins, oos = px[:"2019-12-31"], px["2020-01-01":]

warm = ins.iloc[:200]                    # compile numba once on a small slice, so the timing below is the backtest
wf, ws = vbt.MA.run_combs(warm, window=[2, 3, 4], r=2)
vbt.Portfolio.from_signals(warm, wf.ma_crossed_above(ws), wf.ma_crossed_below(ws), freq="1D").sharpe_ratio()

# 1. all 4,851 pairs at once, fast window below slow window, 5 bp a trade
t0 = time.perf_counter()
fast, slow = vbt.MA.run_combs(ins, window=np.arange(2, 101), r=2, short_names=["fast", "slow"])
entries, exits = fast.ma_crossed_above(slow), fast.ma_crossed_below(slow)
pf = vbt.Portfolio.from_signals(ins, entries, exits, fees=0.0005, freq="1D")
sharpe = pf.sharpe_ratio()
secs = time.perf_counter() - t0
print(f"{sharpe.size:,} backtests on {len(ins):,} days in {secs:.1f} s")

# 2. best pair in sample, then the same pair out of sample against buy and hold
f, s = sharpe.idxmax()
print(f"in sample  2010-2019: best fast {f}, slow {s}, Sharpe {sharpe.max():.2f}, "
      f"median of all {sharpe.median():.2f}, buy and hold {vbt.Portfolio.from_holding(ins, freq='1D').sharpe_ratio():.2f}")
mf, ms = vbt.MA.run(px, f), vbt.MA.run(px, s)                     # full history so the windows are warm on day 1
oos_pf = vbt.Portfolio.from_signals(oos, mf.ma_crossed_above(ms)[oos.index], mf.ma_crossed_below(ms)[oos.index],
                                    fees=0.0005, freq="1D")
hold = vbt.Portfolio.from_holding(oos, freq="1D")
print(f"out of sample {oos.index[0]:%Y}-{oos.index[-1]:%Y}: pair {f}/{s} Sharpe {oos_pf.sharpe_ratio():.2f}, "
      f"return {oos_pf.total_return():.0%}, max drawdown {oos_pf.max_drawdown():.0%} | "
      f"buy and hold Sharpe {hold.sharpe_ratio():.2f}, return {hold.total_return():.0%}, "
      f"max drawdown {hold.max_drawdown():.0%}")

PAPER_PORTS = (7497, 4002)                         # TWS paper, IB Gateway paper; live ports are refused


def connect(port=7497, client_id=11):
    if port not in PAPER_PORTS:
        raise SystemExit(f"port {port} is not a paper port")
    ib = IB()
    ib.connect("127.0.0.1", port, clientId=client_id)
    if not ib.managedAccounts()[0].startswith("DU"):           # paper account ids start with DU
        ib.disconnect()
        raise SystemExit("not a paper account")
    return ib

def send(ib, contract, side, qty, limit=None, wait_s=10):
    """Place the order, wait on executed quantity, return the average fill price (None if unfilled)."""
    trade = ib.placeOrder(contract, LimitOrder(side, qty, limit) if limit else MarketOrder(side, qty))
    for _ in range(wait_s * 2):                    # wait on fills, not on isDone(): a parent order can
        if trade.filled() >= qty:                  # report Cancelled while its executions are correct
            break
        ib.sleep(0.5)
    done = sum(f.execution.shares for f in trade.fills)
    return sum(f.execution.shares * f.execution.price for f in trade.fills) / done if done else None

def route(signal_fn, symbol="SPY", qty=QTY, broker="dry", max_notional=50_000, log=LOG):
    side, ref = signal_fn(symbol)                                       # 4. route, skeleton of route.py
    if side is None or qty * ref > max_notional:                        # risk check before any broker call
        return None
    if broker == "ibkr":
        from ib_async import Stock
        from ibkr_engine import connect, send
        ib = connect()
        contract = Stock(symbol, "SMART", "USD")
        ib.qualifyContracts(contract)
        fill = send(ib, contract, side, qty)                            # 5. fill, read from the executions
        ib.disconnect()
    else:
        fill = ref                                                      # dry run: fill at the reference price
    row = {"t": datetime.now(timezone.utc).isoformat(timespec="seconds"), "broker": broker, "symbol": symbol,
           "side": side, "qty": qty, "signal_px": round(ref, 2), "fill_px": fill and round(fill, 2),
           "slip_bps": fill and round(1e4 * (fill - ref) / ref, 1)}     # signed, compare with the backtest weekly
    os.makedirs(os.path.dirname(log), exist_ok=True)
    with open(log, "a") as f:
        f.write(json.dumps(row) + "\n")
    return row


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--broker", default="dry", choices=["dry", "ibkr"])
    broker = ap.parse_args().broker

    px = data("SPY")
    long = rule(px)
    pf = backtest(px, long)
    print(f"backtest {px.index[0]:%Y-%m-%d} to {px.index[-1]:%Y-%m-%d}: Sharpe {pf.sharpe_ratio():.2f}, "
          f"return {pf.total_return():.0%}, max drawdown {pf.max_drawdown():.0%}, {pf.trades.count()} trades")

    def ma_signal(symbol):                                              # target today minus what is held
        gap = (QTY if long.iloc[-1] else 0) - held(symbol, broker)
        return ("BUY" if gap > 0 else "SELL" if gap < 0 else None), float(px.iloc[-1])

    print(f"rule on {px.index[-1]:%Y-%m-%d}: {'long' if long.iloc[-1] else 'flat'} ({FAST}/{SLOW} day MA)")
    print("order:", route(ma_signal, broker=broker))

backtest 2015-01-02 to 2026-10-16: Sharpe 0.64, return 104%, max drawdown -32%, 13 trades
rule on 2026-10-16: long (43/79 day MA)
order: {'t': '2026-10-19T11:31:34+00:00', 'broker': 'dry', 'symbol': 'SPY', 'side': 'BUY', 'qty': 10, 'signal_px': 763.25, 'fill_px': 763.25, 'slip_bps': 0.0}