from flask import Flask, request, jsonify
import random
import time
from datetime import datetime
from smartapi import SmartConnect
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
# Function to 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
# Simulate live price fetch (replace with actual API call for live market data)
def fetch_live_price(obj, tradingsymbol, token):
try:
live_price = random.uniform(195, 210) # Simulated price for demonstration
return live_price
except Exception as e:
print(f"Error fetching live price: {e}")
return None
# Calculate trailing stop loss
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
# 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
# Function to color the profit/loss text
def colorize_text(value):
if value >= 0:
return f"\033[92m{value}\033[0m" # Green for profit
else:
return f"\033[91m{value}\033[0m" # Red for loss
# Function to calculate Buy Entry for next candle
def calculate_buy_entry_for_next_candle(current_open, current_close):
"""
Calculates the center of the current candle's body and
adds an offset of 0.2 for the next buy entry.
"""
center = (current_open + current_close) / 2 # Candle body center
entry_price = center + 0.2 # Center + 0.2 for next candle entry
return entry_price
# Main strategy function
def nifty_banknifty_strategy(obj, tradingsymbol, token, qty, entry_price):
levels = [0.03, 0.04, 0.05, 0.1, 0.2, 0.3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Fetch current candle data
current_open, current_close = fetch_live_price(obj, tradingsymbol, token), fetch_live_price(obj, tradingsymbol, token)
print(f"Current Candle Open: {current_open}, Close: {current_close}")
# Calculate the buy entry for the next candle
next_entry_price = calculate_buy_entry_for_next_candle(current_open, current_close)
print(f"Calculated Next Candle Entry Price: {next_entry_price}")
# Simulate next candle open price
next_open_price = random.uniform(200, 210) # Simulated next open price
print(f"Next Candle Open Price: {next_open_price}")
# If next candle's open price is greater than the entry price, place the order
if next_open_price >= next_entry_price:
print(f"Placing limit order at next candle's open price: {next_open_price}")
order = place_limit_order(obj, tradingsymbol, token, qty, next_entry_price)
if not order:
return {"status": "error", "message": "Order placement failed."}
# Proceed with the rest of the strategy (stop loss, trailing stop, etc.)
entry_time = datetime.now()
print(f"Entry time: {entry_time}")
price_history = pd.DataFrame(columns=["time", "open", "high", "low", "close", "stop_loss"])
stop_loss = next_entry_price - 0.1
print(f"Initial Stop Loss: {stop_loss}")
while True:
current_price = fetch_live_price(obj, tradingsymbol, token)
if current_price is None:
return {"status": "error", "message": "Price fetch failed."}
profit_loss = (current_price - next_entry_price) * qty
print(f"Current Price: {current_price}, Profit/Loss: {colorize_text(profit_loss)}")
price_history = price_history.append(
{"time": datetime.now(), "open": current_price, "high": current_price,
"low": current_price, "close": current_price, "stop_loss": stop_loss},
ignore_index=True,
)
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}")
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)
price_history.to_csv("price_history.csv", index=False)
return {"status": "success", "message": f"Stop Loss hit at price {current_price}. Profit/Loss: {colorize_text(profit_loss)}"}
elapsed_time = (datetime.now() - entry_time).total_seconds()
if elapsed_time >= 90:
print(f"1.5 minutes elapsed. Exiting trade.")
exit_trade(obj, order["orderid"], current_price, tradingsymbol, token, qty)
price_history.to_csv("price_history.csv", index=False)
return {"status": "success", "message": f"Exited at price {current_price} after 1.5 minutes. Profit/Loss: {colorize_text(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")
entry_price = data.get("entry_price")
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, entry_price)
return jsonify(result)
if __name__ == '__main__':
app.run(debug=True)
No comments