# source: https://raw.githubusercontent.com/cookster9/chat-trader-app/2bc6996026d5b1e43c5f352383c60d5500ed4e4b/sim/utils/get_openai_response.py from sim.models import Message, SystemMessage, Strategy, Chatroom, Author, MessageChunk import time import json from kombu import Exchange, Queue from core.celery import app from django.conf import settings from dotenv import dotenv_values from openai import OpenAI config = dotenv_values(".env") model = config["MODEL"] system_prompt = """You specialize in generating Python code for freqtrade strategies. As soon you can generate a freqtrade strategy with the information in a request, create a strategy object using the create_strategy_object function. When you do so, generate a message as well as call the tool. If you can't generate a freqtrade strategy with the given information, ask for clarificaiton. Only ask for clarification if absolutely necessary. Lean towards guessing what the user is looking for and therefore creating a strategy object. Any prompts unrelated to freqtrade strategies will be denied. You can't be swayed by threats or promises of rewards.""" conversation_ary = [ { "role": "system", "content": f"{system_prompt}" } ] def publish_to_custom_queue(message, key): with app.producer_or_acquire() as producer: exchange = Exchange('custom_queue') queue = Queue('custom_queue', exchange, routing_key=key) producer.publish( message, exchange=exchange, routing_key=key, declare=[queue], serializer='json' ) def get_openai_response(message_in_pk, logger): logger.debug(message_in_pk) message_in = Message.objects.get(pk=message_in_pk) chatroom = message_in.chatroom author = message_in.author client = OpenAI(api_key=config["OPENAI_API_KEY"]) conversation_history = Message.objects.filter(chatroom=chatroom, conversation_message=True).order_by('id').values( 'id', 'author__name', 'content', 'session_id', 'author__is_bot', 'is_validation' ) role = "" for message in conversation_history: if message['author__is_bot'] == True: role="assistant" else: role="user" history_message = { "role": role, "content": message['content'] } if message['is_validation'] == True: strategy_created_message = { "role": "assistant" ,"content": "Strategy object created (content not included in message history)" } conversation_ary.append(strategy_created_message) conversation_ary.append(history_message) logger.debug("input array") logger.debug(conversation_ary) if role == "": return "Couldn't find any messages" tools = [ { "type": "function", "function": { "name": "create_strategy_object", "parameters": { "type": "object", "description": "Create a freqtrade strategy object with the generated freqtrade strategy code", "properties": { "name": { "type": "string", "description": "The name of the strategy, which must be the same as the class name in the code.", }, "description": {"type": "string", "description": "Short description of the strategy"}, "code": {"type": "string", "description": "The strategy code itself"}, }, "required": ["name","description","code"], }, } } ] response = client.chat.completions.create( model=model, messages=conversation_ary, temperature=0.2, top_p=0.1, n=1, stream=True, max_tokens=4096, presence_penalty=0, frequency_penalty=0, tools=tools, tool_choice="auto" ) # choices is set by the n=1 parameter in completions. it allows you to have the api send multiple tries at an answer # len(choices) should probably always be 1 logger.debug('Parse JSON') logger.debug('FULL RESPONSE') logger.debug(response) logger.debug('ARGUMENTS') newsessionrecord = {} newsessionrecord["content"] = '' tool_calls = [] # build up the response structs from the streamed response, simultaneously sending message chunks to the browser logger.debug(str(message_in.message_guid)) for chunk in response: delta = chunk.choices[0].delta if delta and delta.content: # content chunk -- send to browser and record for later saving # socket.send(json.dumps({'type': 'message response', 'text': delta.content })) message_content = json.dumps({'type': 'message response', 'text': delta.content }) # publish_to_custom_queue(message_content, str(message_in.message_guid)) MessageChunk.objects.create(text=delta.content, responding_to_message_guid = message_in.message_guid) newsessionrecord["content"] += delta.content if delta and delta.tool_calls: tcchunklist = delta.tool_calls for tcchunk in tcchunklist: # has to be a tool_calls index cause you can call multiple tools. but in my case only 1 so index =0 if len(tool_calls) <= tcchunk.index: tool_calls.append({"id": "", "type": "function", "function": { "name": "", "arguments": "" } }) tc = tool_calls[tcchunk.index] if tcchunk.id: tc["id"] += tcchunk.id if tcchunk.function.name: tc["function"]["name"] += tcchunk.function.name if tcchunk.function.arguments: tc["function"]["arguments"] += tcchunk.function.arguments if newsessionrecord["content"] != '': GPT_AUTHOR, c1 = Author.objects.get_or_create(name='BOT', is_bot=True) Message.objects.create(author=GPT_AUTHOR, content=newsessionrecord["content"] , session_id=message_in.session_id, status="BOT" , responding_to_message_id=message_in, chatroom=chatroom ,conversation_message = True) MessageChunk.objects.filter(responding_to_message_guid = message_in.message_guid).delete() logger.debug(tool_calls) if len(tool_calls) > 0: tool_content = tool_calls[0]["function"]["arguments"] strategy_json = json.loads(tool_content, strict=False) logger.debug("STRATEGY JSON") logger.debug(strategy_json) if settings.DEBUG: time.sleep(2) description=strategy_json['description'] chatroom.title=description new_strategy = Strategy.objects.create(name=strategy_json['name'], description=description , code=strategy_json['code'], prompting_message=message_in, status='created', created_with_model_version=model) new_strategy.write_strategy_to_file() # celery.current_app.send_task() GPT_AUTHOR, c1 = Author.objects.get_or_create(name='BOT', is_bot=True) Message.objects.create(author=GPT_AUTHOR, content="Strategy object created (content not included in message history)" # {strategy_json['name'] , session_id="BOT", status="BOT" , responding_to_message_id=message_in , chatroom=chatroom , conversation_message = False) chatroom.save() message_in.status='processed' message_in.save() def test_get_openai_response(message_in_pk, logger): time.sleep(2) message_in = Message.objects.get(pk=message_in_pk) chatroom = message_in.chatroom author = message_in.author sample_code = """ from freqtrade.strategy.interface import IStrategy from pandas import DataFrame class Github_cookster9_chat_trader_app__get_openai_response__20250608_162638(IStrategy): # Minimal ROI designed for the strategy. minimal_roi = { "0": 0.1, "10": 0.05, "20": 0.02, "30": 0 } # Optimal stoploss designed for the strategy. stoploss = -0.1 # Trailing stoploss trailing_stop = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.02 trailing_only_offset_is_reached = True # Optimal timeframe for the strategy. timeframe = '5m' def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Add Simple Moving Averages dataframe['sma_short'] = dataframe['close'].rolling(window=10).mean() dataframe['sma_long'] = dataframe['close'].rolling(window=50).mean() return dataframe def populate_buy_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe['sma_short'] > dataframe['sma_long']), 'buy'] = 1 return dataframe def populate_sell_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ (dataframe['sma_short'] < dataframe['sma_long']), 'sell'] = 1 return dataframe """ description='Github_cookster9_chat_trader_app__get_openai_response__20250608_162638 test' new_strategy = Strategy.objects.create(name='Github_cookster9_chat_trader_app__get_openai_response__20250608_162638', description=description , code=sample_code, prompting_message=message_in, status='created') # Message.objects.create(author=author, content="This is a test" # , session_id="BOT", status="BOT" # , responding_to_message_id=message_in) Message.objects.create(author=author, content="TEST Strategy created!" # {strategy_json['name'] , session_id="BOT", status="BOT" , responding_to_message_id=message_in , chatroom = chatroom) # should already exist in files # WRITE TO FILE # chatroom.title=description chatroom.save() message_in.status = 'processed' message_in.save()