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:
# Example: Replace with actual SmartAPI live price fetching method
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):
"""
Adjust stop loss based on the current price and pre-defined levels.
"""
stop_loss = entry_price # Initial stop loss is set to 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() # Select the highest level below current price
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:
# Green color for profit
return f"\033[92m{value}\033[0m" # Green
else:
# Red color for loss
return f"\033[91m{value}\033[0m" # Red
# Main strategy function
def nifty_banknifty_strategy(obj, tradingsymbol, token, qty, entry_price):
# Define levels for trailing stop loss
levels = [0.03, 0.04, 0.05, 0.1, 0.2, 0.3, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
# Place the limit order
order = place_limit_order(obj, tradingsymbol, token, qty, entry_price)
if not order:
return {"status": "error", "message": "Order placement failed."}
# Log entry time
entry_time = datetime.now()
print(f"Entry time: {entry_time}")
# Create a DataFrame to track price data
price_history = pd.DataFrame(columns=["time", "open", "high", "low", "close", "stop_loss"])
# Set initial stop loss
stop_loss = entry_price - 0.1 # Fixed stop loss = Entry Price - 0.1
print(f"Initial Stop Loss: {stop_loss}")
while True:
# Fetch current price
current_price = fetch_live_price(obj, tradingsymbol, token)
if current_price is None:
return {"status": "error", "message": "Price fetch failed."}
# Calculate real-time profit/loss
profit_loss = (current_price - entry_price) * qty
print(f"Current Price: {current_price}, Profit/Loss: {colorize_text(profit_loss)}")
# Add current price and stop loss to the DataFrame
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,
)
# Update trailing stop loss
trailing_stop_loss = calculate_trailing_stop_loss(entry_price, current_price, levels)
if trailing_stop_loss > stop_loss: # Update only if trailing SL is higher
stop_loss = trailing_stop_loss
print(f"Trailing Stop Loss updated to: {stop_loss}")
# Check for stop loss 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)
# Save the price history to a CSV file for analysis
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)}"}
# Check for 1.5-minute exit
elapsed_time = (datetime.now() - entry_time).total_seconds()
if elapsed_time >= 90: # 90 seconds = 1.5 minutes
print(f"1.5 minutes elapsed. Exiting trade.")
exit_trade(obj, order["orderid"], current_price, tradingsymbol, token, qty)
# Save the price history to a CSV file for analysis
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) # Wait for 1 second before next price check
@app.route('/start_trading', methods=['POST'])
def start_trading():
# Get parameters from the request
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")
# Connect to the API
obj = connect_to_api(api_key, access_token)
if obj is None:
return jsonify({"status": "error", "message": "API connection failed."})
# Start the strategy
result = nifty_banknifty_strategy(obj, tradingsymbol, token, qty, entry_price)
return jsonify(result)
if __name__ == '__main__':
app.run(debug=True)
No comments