from flask import Flask, request, jsonify
from smartapi import SmartConnect
from datetime import datetime
import time
import pandas as pd
import numpy as np
app = Flask(__name__)
# Connect to SmartAPI
def connect_to_api(api_key, access_token):
try:
obj = SmartConnect(api_key=api_key, access_token=access_token)
print("Connected to SmartAPI successfully.")
return obj
except Exception as e:
print(f"Error connecting to SmartAPI: {e}")
return None
# Fetch 5-minute candle data
def fetch_5min_candle_data(obj, tradingsymbol, interval="5minute"):
try:
data = obj.getCandleData(
exchange="NFO",
tradingsymbol=tradingsymbol,
interval=interval,
fromdate="2023-01-01 09:15", # Replace with dynamic start date
todate=datetime.now().strftime("%Y-%m-%d %H:%M")
)
return data["data"]
except Exception as e:
print(f"Error fetching 5-minute candle data: {e}")
return None
# Place a limit buy order
def place_limit_order(obj, tradingsymbol, token, qty, entry_price):
try:
order = obj.place_order(
exchange="NFO",
tradingsymbol=tradingsymbol,
symboltoken=token,
transactiontype="BUY",
qty=qty,
price=entry_price,
producttype="INTRADAY",
ordertype="LIMIT"
)
print(f"Limit order placed at {entry_price}, Order ID: {order['orderid']}")
return order
except Exception as e:
print(f"Error placing limit order: {e}")
return None
# Exit trade
def exit_trade(obj, order_id, exit_price, tradingsymbol, token, qty):
try:
exit_order = obj.place_order(
exchange="NFO",
tradingsymbol=tradingsymbol,
symboltoken=token,
transactiontype="SELL",
qty=qty,
price=exit_price,
producttype="INTRADAY",
ordertype="LIMIT"
)
print(f"Exit order placed at {exit_price}, Order ID: {exit_order['orderid']}")
return exit_order
except Exception as e:
print(f"Error placing exit order: {e}")
return None
# Trailing stop loss calculation
def calculate_trailing_stop_loss(entry_price, current_price, levels):
stop_loss = entry_price
possible_levels = np.array(levels) + entry_price
valid_levels = possible_levels[possible_levels <= current_price]
if valid_levels.size > 0:
stop_loss = valid_levels.max()
return stop_loss
# Calculate buy entry price for the next candle
def calculate_buy_entry_for_next_candle(current_open, current_close):
center = (current_open + current_close) / 2
entry_price = center + 0.2 # Center + offset
return entry_price
# Main strategy function
def nifty_banknifty_strategy(obj, tradingsymbol, token, qty):
levels = [0.03, 0.04, 0.05, 0.1, 0.2, 0.3, 1, 2, 3, 4, 5]
while True:
# Fetch the latest 5-minute candle
candle_data = fetch_5min_candle_data(obj, tradingsymbol)
if not candle_data:
print("No candle data available.")
continue
current_candle = candle_data[-1] # Latest candle
current_open = current_candle[1]
current_close = current_candle[4]
print(f"5-Minute Candle - Open: {current_open}, Close: {current_close}")
# Calculate the next entry price
next_entry_price = calculate_buy_entry_for_next_candle(current_open, current_close)
print(f"Calculated Next Candle Entry Price: {next_entry_price}")
# Place the limit order if conditions match
if current_close >= next_entry_price:
print(f"Placing order at: {next_entry_price}")
order = place_limit_order(obj, tradingsymbol, token, qty, next_entry_price)
if not order:
return {"status": "error", "message": "Order placement failed."}
# Start trailing stop loss and exit logic
entry_time = datetime.now()
stop_loss = next_entry_price - 0.1
print(f"Initial Stop Loss: {stop_loss}")
while True:
# Fetch the latest candle data
candle_data = fetch_5min_candle_data(obj, tradingsymbol)
current_candle = candle_data[-1]
current_price = current_candle[4] # Close price
# Calculate profit/loss
profit_loss = (current_price - next_entry_price) * qty
print(f"Current Price: {current_price}, Profit/Loss: {profit_loss}")
# Update trailing stop loss
trailing_stop_loss = calculate_trailing_stop_loss(next_entry_price, current_price, levels)
if trailing_stop_loss > stop_loss:
stop_loss = trailing_stop_loss
print(f"Trailing Stop Loss updated to: {stop_loss}")
# Exit trade if stop loss is hit
if current_price <= stop_loss:
print(f"Stop Loss hit at {current_price}. Exiting trade.")
exit_trade(obj, order["orderid"], current_price, tradingsymbol, token, qty)
return {"status": "success", "message": f"Stop Loss hit at price {current_price}. Profit/Loss: {profit_loss}"}
# Exit after 5 minutes
elapsed_time = (datetime.now() - entry_time).total_seconds()
if elapsed_time >= 300:
print(f"5 minutes elapsed. Exiting trade.")
exit_trade(obj, order["orderid"], current_price, tradingsymbol, token, qty)
return {"status": "success", "message": f"Exited at price {current_price} after 5 minutes. Profit/Loss: {profit_loss}"}
time.sleep(1)
@app.route('/start_trading', methods=['POST'])
def start_trading():
data = request.json
api_key = data.get("api_key")
access_token = data.get("access_token")
tradingsymbol = data.get("tradingsymbol")
token = data.get("token")
qty = data.get("qty")
obj = connect_to_api(api_key, access_token)
if obj is None:
return jsonify({"status": "error", "message": "API connection failed."})
result = nifty_banknifty_strategy(obj, tradingsymbol, token, qty)
return jsonify(result)
if __name__ == '__main__':
app.run(debug=True)
No comments