How to write a basic Python script using kiteconnect

mkdir kite-project && cd kite-project
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install kiteconnect python-dotenv
pip freeze > requirements.txt

"""auth.py, Kite Connect daily login flow."""

import os
import json
from pathlib import Path
from dotenv import load_dotenv
from kiteconnect import KiteConnect

load_dotenv()

API_KEY = os.environ["KITE_API_KEY"]
API_SECRET = os.environ["KITE_API_SECRET"]
TOKEN_FILE = Path("kite_token.json")


def get_kite_client() -> KiteConnect:
    """Return an authenticated KiteConnect instance for today's session."""
    kite = KiteConnect(api_key=API_KEY)

    if TOKEN_FILE.exists():
        data = json.loads(TOKEN_FILE.read_text())
        access_token = data.get("access_token")
        if access_token:
            kite.set_access_token(access_token)
            # Quick validation call; raises TokenException if stale
            try:
                kite.profile()
                return kite
            except Exception:
                pass  # Token stale; fall through to re-login

    # Interactive daily login
    print("Open this URL in a browser:")
    print(kite.login_url())
    request_token = input("Paste the request_token from the redirect URL: ").strip()

    session = kite.generate_session(request_token, api_secret=API_SECRET)
    kite.set_access_token(session["access_token"])
    TOKEN_FILE.write_text(json.dumps({"access_token": session["access_token"]}))
    print("Authenticated successfully.")
    return kite


if __name__ == "__main__":
    kite = get_kite_client()
    print("Logged in as:", kite.profile()["user_name"])

"""main.py, Portfolio, positions, and market data queries."""

from auth import get_kite_client

kite = get_kite_client()

# --- Profile ---
profile = kite.profile()
print(f"\nLogged in as: {profile['user_name']} ({profile['user_id']})")
print(f"Broker: {profile['broker']}")

# --- Account margins ---
margins = kite.margins()
print(f"\nEquity available margin: {margins['equity']['available']['live_balance']}")
print(f"Commodity available margin: {margins['commodity']['available']['live_balance']}")

# --- Holdings (delivery positions) ---
holdings = kite.holdings()
print(f"\nHoldings ({len(holdings)} positions):")
for h in holdings:
    print(f"  {h['tradingsymbol']:20s}  qty={h['quantity']:5d}  avg={h['average_price']:10.2f}  ltp={h['last_price']:10.2f}")

# --- Intraday positions ---
positions = kite.positions()
day = positions["day"]
net = positions["net"]
print(f"\nIntraday positions (day): {len(day)}")
for p in day:
    print(f"  {p['tradingsymbol']:20s}  qty={p['quantity']:5d}  pnl={p['pnl']:10.2f}")

# --- Live quotes ---
symbols = ["NSE:RELIANCE", "NSE:INFY", "NSE:HDFCBANK"]
quotes = kite.quote(symbols)
print("\nLive quotes:")
for sym, q in quotes.items():
    print(f"  {sym:25s}  ltp={q['last_price']:10.2f}  volume={q['volume']:10d}")

# --- Historical OHLC (last 5 days) ---
import datetime
instruments = kite.instruments("NSE")
reliance_token = next(i["instrument_token"] for i in instruments if i["tradingsymbol"] == "RELIANCE")

to_date = datetime.date.today()
from_date = to_date - datetime.timedelta(days=7)

candles = kite.historical_data(
    instrument_token=reliance_token,
    from_date=from_date,
    to_date=to_date,
    interval="day",
)
print(f"\nRELIANCE daily OHLC (last {len(candles)} sessions):")
for c in candles[-5:]:
    print(f"  {c['date'].date()}  O={c['open']:.2f}  H={c['high']:.2f}  L={c['low']:.2f}  C={c['close']:.2f}  V={c['volume']}")

"""orders.py, Safe wrappers around kite.place_order()."""

from kiteconnect import KiteConnect
from kiteconnect.exceptions import InputException, OrderException, NetworkException


def place_market_order(
    kite: KiteConnect,
    exchange: str,
    symbol: str,
    transaction_type: str,
    quantity: int,
    product: str,
) -> str | None:
    """Place a market order and return the order_id, or None on failure."""
    if quantity <= 0:
        raise ValueError(f"Quantity must be positive, got {quantity}")
    if transaction_type not in ("BUY", "SELL"):
        raise ValueError(f"transaction_type must be BUY or SELL, got {transaction_type}")

    try:
        order_id = kite.place_order(
            variety=kite.VARIETY_REGULAR,
            exchange=exchange,
            tradingsymbol=symbol,
            transaction_type=transaction_type,
            quantity=quantity,
            product=product,
            order_type=kite.ORDER_TYPE_MARKET,
        )
        print(f"Order placed: {order_id}")
        return order_id
    except InputException as e:
        print(f"Input error: {e}")
    except OrderException as e:
        print(f"Order rejected: {e}")
    except NetworkException as e:
        print(f"Network error: {e}")
    return None


def get_order_status(kite: KiteConnect, order_id: str) -> dict | None:
    """Return the latest status dict for an order_id."""
    orders = kite.orders()
    for o in orders:
        if o["order_id"] == order_id:
            return o
    return None

from orders import place_market_order, get_order_status

# Uncomment to place a real order (ensure you have sufficient margin)
# order_id = place_market_order(kite, "NSE", "RELIANCE", "BUY", 1, "CNC")
# if order_id:
#     status = get_order_status(kite, order_id)
#     print("Order status:", status["status"])

# With virtual environment active:
python auth.py     # authenticate once; token saved to kite_token.json
python main.py     # fetch portfolio and market data