125 lines
4.6 KiB
Python
125 lines
4.6 KiB
Python
import os
|
|
import webbrowser
|
|
from dotenv import load_dotenv
|
|
from fyers_api import fyersModel, accessToken
|
|
from loguru import logger
|
|
|
|
class FyersBroker:
|
|
"""
|
|
Handles all interactions with the Fyers API.
|
|
"""
|
|
def __init__(self, settings: dict):
|
|
"""
|
|
Initializes the FyersBroker.
|
|
|
|
Args:
|
|
settings (dict): Application settings containing API credentials.
|
|
"""
|
|
load_dotenv(dotenv_path=settings['credentials_path'])
|
|
self.client_id = os.getenv('FYERS_CLIENT_ID')
|
|
self.secret_key = os.getenv('FYERS_SECRET_KEY')
|
|
self.redirect_uri = os.getenv('FYERS_REDIRECT_URI')
|
|
self.access_token_path = settings['access_token_path']
|
|
|
|
self.fyers = None
|
|
self.access_token = self._load_access_token()
|
|
|
|
if not self.access_token:
|
|
self._generate_new_access_token()
|
|
|
|
self._initialize_fyers_model()
|
|
|
|
def _load_access_token(self) -> str | None:
|
|
"""Loads access token from the specified file."""
|
|
try:
|
|
if os.path.exists(self.access_token_path):
|
|
with open(self.access_token_path, 'r') as f:
|
|
return f.read().strip()
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error loading access token: {e}")
|
|
return None
|
|
|
|
def _save_access_token(self, token: str):
|
|
"""Saves access token to the specified file."""
|
|
try:
|
|
with open(self.access_token_path, 'w') as f:
|
|
f.write(token)
|
|
logger.info("Access token saved successfully.")
|
|
except Exception as e:
|
|
logger.error(f"Error saving access token: {e}")
|
|
|
|
def _generate_new_access_token(self):
|
|
"""Generates a new access token using the auth code flow."""
|
|
session = accessToken.SessionModel(
|
|
client_id=self.client_id,
|
|
secret_key=self.secret_key,
|
|
redirect_uri=self.redirect_uri,
|
|
response_type='code',
|
|
grant_type='authorization_code'
|
|
)
|
|
|
|
try:
|
|
response = session.generate_authcode()
|
|
logger.info(f"Auth code generation response: {response}")
|
|
print(f"Login URL: {response}")
|
|
webbrowser.open(response)
|
|
|
|
auth_code = input("Enter the auth code from the redirected URL: ")
|
|
session.set_token(auth_code)
|
|
|
|
access_token_response = session.generate_token()
|
|
self.access_token = access_token_response['access_token']
|
|
self._save_access_token(self.access_token)
|
|
logger.info("New access token generated and saved.")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error generating new access token: {e}")
|
|
self.access_token = None
|
|
|
|
def _initialize_fyers_model(self):
|
|
"""Initializes the FyersModel with the access token."""
|
|
if self.access_token:
|
|
try:
|
|
self.fyers = fyersModel.FyersModel(
|
|
client_id=self.client_id,
|
|
is_async=True,
|
|
token=self.access_token,
|
|
log_path=os.path.join(os.path.dirname(__file__), '..' , 'logs')
|
|
)
|
|
logger.info("FyersModel initialized successfully.")
|
|
except Exception as e:
|
|
logger.error(f"Error initializing FyersModel: {e}")
|
|
self.fyers = None
|
|
else:
|
|
logger.warning("FyersModel could not be initialized. Access token is missing.")
|
|
|
|
async def get_profile(self):
|
|
"""Fetches user profile details."""
|
|
if not self.fyers:
|
|
return {"error": "FyersModel not initialized."}
|
|
try:
|
|
response = await self.fyers.get_profile()
|
|
if response['s'] == 'ok':
|
|
return response['data']
|
|
else:
|
|
logger.error(f"Error fetching profile: {response['message']}")
|
|
return {"error": response['message']}
|
|
except Exception as e:
|
|
logger.error(f"Exception fetching profile: {e}")
|
|
return {"error": str(e)}
|
|
|
|
async def get_funds(self):
|
|
"""Fetches user funds details."""
|
|
if not self.fyers:
|
|
return {"error": "FyersModel not initialized."}
|
|
try:
|
|
response = await self.fyers.get_funds()
|
|
if response['s'] == 'ok':
|
|
return response['fund_limit']
|
|
else:
|
|
logger.error(f"Error fetching funds: {response['message']}")
|
|
return {"error": response['message']}
|
|
except Exception as e:
|
|
logger.error(f"Exception fetching funds: {e}")
|
|
return {"error": str(e)} |