merge all commit

* Adjust prompt and fix cases
* adjust summarizeTask & learn prompts;
* fix typos & drop duplicate task method;

* adjust learn prompts;

.gitignore

@@ -22,6 +22,7 @@ dist/
 qlib/VERSION.txt
 qlib/data/_libs/expanding.cpp
 qlib/data/_libs/rolling.cpp
+qlib/finco/prompt_cache.json
 examples/estimator/estimator_example/
 examples/rl/data/
 examples/rl/checkpoints/

qlib/contrib/analyzer.py

@@ -0,0 +1,111 @@
+import logging
+import matplotlib.pyplot as plt
+from pathlib import Path
+import numpy as np
+
+from ..log import get_module_logger
+from ..contrib.eva.alpha import calc_ic, calc_long_short_return, calc_long_short_prec
+
+logger = get_module_logger("analysis", logging.INFO)
+
+
+class AnalyzerTemp:
+    def __init__(self, recorder, output_dir=None, **kwargs):
+        self.recorder = recorder
+        self.output_dir = Path(output_dir) if output_dir else "./"
+
+    def load(self, name: str):
+        """
+        It behaves the same as self.recorder.load_object.
+        But it is an easier interface because users don't have to care about `get_path` and `artifact_path`
+
+        Parameters
+        ----------
+        name : str
+            the name for the file to be load.
+
+        Return
+        ------
+        The stored records.
+        """
+        return self.recorder.load_object(name)
+
+    def analyse(self, **kwargs):
+        """
+        Analyse data index, distribution .etc
+
+        Parameters
+        ----------
+
+
+        Return
+        ------
+        The handled data.
+        """
+        raise NotImplementedError(f"Please implement the `analysis` method.")
+
+
+class HFAnalyzer(AnalyzerTemp):
+    """
+    This is the Signal Analysis class that generates the analysis results such as IC and IR.
+
+    default output image filename is "HFAnalyzerTable.jpeg"
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def analyse(self):
+        pred = self.load("pred.pkl")
+        label = self.load("label.pkl")
+
+        long_pre, short_pre = calc_long_short_prec(pred.iloc[:, 0], label.iloc[:, 0], is_alpha=True)
+        ic, ric = calc_ic(pred.iloc[:, 0], label.iloc[:, 0])
+        metrics = {
+            "IC": ic.mean(),
+            "ICIR": ic.mean() / ic.std(),
+            "Rank IC": ric.mean(),
+            "Rank ICIR": ric.mean() / ric.std(),
+            "Long precision": long_pre.mean(),
+            "Short precision": short_pre.mean(),
+        }
+
+        long_short_r, long_avg_r = calc_long_short_return(pred.iloc[:, 0], label.iloc[:, 0])
+        metrics.update(
+            {
+                "Long-Short Average Return": long_short_r.mean(),
+                "Long-Short Average Sharpe": long_short_r.mean() / long_short_r.std(),
+            }
+        )
+
+        table = [[k, v] for (k, v) in metrics.items()]
+        plt.table(cellText=table, loc="center")
+        plt.axis("off")
+        plt.savefig(self.output_dir.joinpath("HFAnalyzerTable.jpeg"))
+        plt.clf()
+
+        plt.scatter(np.arange(0, len(pred)), pred.iloc[:, 0])
+        plt.scatter(np.arange(0, len(label)), label.iloc[:, 0])
+        plt.title("HFAnalyzer")
+        plt.savefig(self.output_dir.joinpath("HFAnalyzer.jpeg"))
+        return "HFAnalyzer.jpeg"
+
+
+class SignalAnalyzer(AnalyzerTemp):
+    """
+    This is the Signal Analysis class that generates the analysis results such as IC and IR.
+
+    default output image filename is "signalAnalysis.jpeg"
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    def analyse(self, dataset=None, **kwargs):
+        label = self.load("label.pkl")
+
+        plt.hist(label)
+        plt.title("SignalAnalyzer")
+        plt.savefig(self.output_dir.joinpath("signalAnalysis.jpeg"))
+
+        return "signalAnalysis.jpeg"

qlib/contrib/data/handler.py

@@ -1,6 +1,8 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
+from typing import Optional
+from qlib.utils.data import update_config
 from ...data.dataset.handler import DataHandlerLP
 from ...data.dataset.processor import Processor
 from ...utils import get_callable_kwargs
@@ -57,12 +59,13 @@ class Alpha360(DataHandlerLP):
         fit_end_time=None,
         filter_pipe=None,
         inst_processors=None,
+        data_loader: Optional[dict] = None,
         **kwargs
     ):
         infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
         learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
 
-        data_loader = {
+        _data_loader = {
             "class": "QlibDataLoader",
             "kwargs": {
                 "config": {
@@ -74,12 +77,14 @@ class Alpha360(DataHandlerLP):
                 "inst_processors": inst_processors,
             },
         }
+        if data_loader is not None:
+            update_config(_data_loader, data_loader)
 
         super().__init__(
             instruments=instruments,
             start_time=start_time,
             end_time=end_time,
-            data_loader=data_loader,
+            data_loader=_data_loader,
             learn_processors=learn_processors,
             infer_processors=infer_processors,
             **kwargs
@@ -153,12 +158,13 @@ class Alpha158(DataHandlerLP):
         process_type=DataHandlerLP.PTYPE_A,
         filter_pipe=None,
         inst_processors=None,
+        data_loader: Optional[dict] = None,
         **kwargs
     ):
         infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
         learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
 
-        data_loader = {
+        _data_loader = {
             "class": "QlibDataLoader",
             "kwargs": {
                 "config": {
@@ -170,11 +176,13 @@ class Alpha158(DataHandlerLP):
                 "inst_processors": inst_processors,
             },
         }
+        if data_loader is not None:
+            update_config(_data_loader, data_loader)
         super().__init__(
             instruments=instruments,
             start_time=start_time,
             end_time=end_time,
-            data_loader=data_loader,
+            data_loader=_data_loader,
             infer_processors=infer_processors,
             learn_processors=learn_processors,
             process_type=process_type,

qlib/finco/.env.example

@@ -0,0 +1,18 @@
+
+OPENAI_API_KEY=your_api_key
+
+# USE_AZURE=True
+# AZURE_API_BASE=your_api_base
+# AZURE_API_VERSION=your_api_version
+
+# use gpt-4 means more token but more wait time
+# MODEL=gpt-4
+# MAX_TOKENS=1600
+# MAX_RETRY=1000
+
+
+MAX_TOKENS=1600
+MAX_RETRY=120
+
+CONTINOUS_MODE=True
+DEBUG_MODE=True

qlib/finco/README.md

@@ -0,0 +1,22 @@
+# This is an experimental branch of "`FI`nancial `CO`pilot of `Qlib`"
+
+## Installation
+
+- To run this module, you need to first install Qlib following the instruction in [install-from-source](/README.md#install-from-source) or follow:
+
+```python
+python -m pip install git+https://github.com/microsoft/qlib.git@finco
+```
+
+- then you need to install other dependencies of finco:
+```python
+python -m pip install pydantic openai python-dotenv
+```
+
+## Quick run
+
+To run this module, you can start the workflow easily with one command:
+
+```sh
+cd qlib/finco; python cli.py "your prompt"
+```

qlib/finco/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+from pathlib import Path
+
+DIRNAME = Path(__file__).absolute().resolve().parent
+
+
+def get_finco_path() -> Path:
+    """
+    return the template path
+    Because the template path is located in the folder. We don't know where it is located. So __file__ for this module will be used.
+    """
+    return DIRNAME

qlib/finco/cli.py

@@ -0,0 +1,15 @@
+import fire
+from qlib.finco.workflow import WorkflowManager
+from dotenv import load_dotenv
+from qlib import auto_init
+
+
+def main(prompt=None):
+    load_dotenv(verbose=True, override=True)
+    wm = WorkflowManager()
+    wm.run(prompt)
+
+
+if __name__ == "__main__":
+    auto_init()
+    fire.Fire(main)

qlib/finco/cli_learn.py

@@ -0,0 +1,15 @@
+import fire
+from qlib.finco.workflow import LearnManager
+from dotenv import load_dotenv
+from qlib import auto_init
+
+
+def main(prompt=None):
+    load_dotenv(verbose=True, override=True)
+    lm = LearnManager()
+    lm.run(prompt)
+
+
+if __name__ == "__main__":
+    auto_init()
+    fire.Fire(main)

qlib/finco/conf.py

@@ -0,0 +1,32 @@
+# TODO: use pydantic for other modules in Qlib
+from pydantic import BaseSettings
+from qlib.finco.utils import SingletonBaseClass
+
+import os
+
+
+class Config(SingletonBaseClass):
+    """
+    This config is for fast demo purpose.
+    Please use BaseSettings insetead in the future
+    """
+
+    def __init__(self):
+        self.use_azure = os.getenv("USE_AZURE") == "True"
+        self.temperature = 0.5 if os.getenv("TEMPERATURE") is None else float(os.getenv("TEMPERATURE"))
+        self.max_tokens = 800 if os.getenv("MAX_TOKENS") is None else int(os.getenv("MAX_TOKENS"))
+
+        self.openai_api_key = os.getenv("OPENAI_API_KEY")
+        self.use_azure = os.getenv("USE_AZURE") == "True"
+        self.azure_api_base = os.getenv("AZURE_API_BASE")
+        self.azure_api_version = os.getenv("AZURE_API_VERSION")
+        self.model = os.getenv("MODEL") or ("gpt-35-turbo" if self.use_azure else "gpt-3.5-turbo")
+
+        self.max_retry = int(os.getenv("MAX_RETRY")) if os.getenv("MAX_RETRY") is not None else None
+
+        self.continuous_mode = (
+            os.getenv("CONTINOUS_MODE") == "True" if os.getenv("CONTINOUS_MODE") is not None else False
+        )
+        self.debug_mode = os.getenv("DEBUG_MODE") == "True" if os.getenv("DEBUG_MODE") is not None else False
+        self.workspace = os.getenv("WORKSPACE") if os.getenv("WORKSPACE") is not None else "./finco_workspace"
+        self.max_past_message_include = int(os.getenv("MAX_PAST_MESSAGE_INCLUDE") or 6) // 2 * 2

qlib/finco/knowledge.py

@@ -0,0 +1,156 @@
+from pathlib import Path
+from jinja2 import Template
+from typing import List
+
+from qlib.workflow import R
+from qlib.finco.log import FinCoLog
+from qlib.finco.llm import APIBackend
+
+
+class Knowledge:
+    """
+    Use to handle knowledge in finCo such as experiment and outside domain information
+    """
+
+    def __init__(self):
+        self.logger = FinCoLog()
+
+    def load(self, **kwargs):
+        """
+        Load knowledge in memory
+
+        Parameters
+        ----------
+
+        Return
+        ------
+        """
+        raise NotImplementedError(f"Please implement the `load` method.")
+
+    def brief(self, **kwargs):
+        """
+        Return a brief summary of knowledge
+
+        Parameters
+        ----------
+
+        Return
+        ------
+        """
+        raise NotImplementedError(f"Please implement the `load` method.")
+
+
+class KnowledgeExperiment(Knowledge):
+    """
+    Handle knowledge from experiments
+    """
+
+    def __init__(self, exp_name, rec_id=None):
+        super().__init__()
+        self.exp_name = exp_name
+        self.exp = None
+        self.recs = []
+
+        self.load(exp_name=exp_name, rec_id=rec_id)
+
+    def load(self, exp_name, rec_id=None):
+        recs = []
+        self.exp = R.get_exp(experiment_name=exp_name)
+        for r in self.exp.list_recorders(rtype=self.exp.RT_L):
+            if rec_id is not None and r.id != rec_id:
+                continue
+            recs.append(r)
+        self.recs.extend(recs)
+
+    def brief(self):
+        docs = []
+        for recorder in self.recs:
+            docs.append({"exp_name": self.exp.name, "record_info": recorder.info,
+                         "config": recorder.load_object("config"),
+                         "context_summary": recorder.load_object("context_summary")})
+
+        return docs
+
+
+class Topic:
+
+    def __init__(self, name: str, describe: Template):
+        self.name = name
+        self.describe = describe
+        self.docs = []
+        self.knowledge = None
+        self.logger = FinCoLog()
+
+    def summarize(self, docs: list):
+        self.logger.info(f"Summarize topic: \nname: {self.name}\ndescribe: {self.describe.module}")
+        prompt_workflow_selection = self.describe.render(docs=docs)
+        response = APIBackend().build_messages_and_create_chat_completion(
+            user_prompt=prompt_workflow_selection
+        )
+
+        self.knowledge = response
+        self.docs = docs
+
+
+class KnowledgeBase:
+    """
+    Load knowledge, offer brief information of knowledge and common handle interfaces
+    """
+
+    def __init__(self, init_path=None, topics: List[Topic] = None):
+        self.logger = FinCoLog()
+        init_path = init_path if init_path else Path.cwd()
+
+        if not init_path.exists():
+            self.logger.warning(f"{init_path} not exist, create empty directory.")
+            Path.mkdir(init_path)
+
+        self.knowledge = self.load(path=init_path)
+
+        # todo: replace list with persistent storage strategy such as ES/pinecone to enable
+        # literal search/semantic search
+        self.docs = self.brief(knowledge=self.knowledge)
+
+        self.topics = topics if topics else []
+
+    def load(self, path) -> List:
+        if isinstance(path, str):
+            path = Path(path)
+
+        knowledge = []
+        path = path if path.name == "mlruns" else path.joinpath("mlruns")
+        R.set_uri(path.as_uri())
+        for exp_name in R.list_experiments():
+            knowledge.append(KnowledgeExperiment(exp_name=exp_name))
+
+        self.logger.plain_info(f"Load knowledge from: {path} finished.")
+        return knowledge
+
+    def update(self, path):
+        # note: only update new knowledge in future
+        knowledge = self.load(path)
+        self.knowledge = knowledge
+        self.docs = self.brief(self.knowledge)
+        self.logger.plain_info(f"Update knowledge finished.")
+
+    def brief(self, knowledge: List[Knowledge]) -> List:
+        docs = []
+        for k in knowledge:
+            docs.extend(k.brief())
+
+        self.logger.plain_info(f"Generate brief knowledge summary finished.")
+        return docs
+
+    def query(self, content: str = None):
+        # todo: query by DSL
+        return self.docs
+
+    def query_topics(self):
+        knowledge_of_topics = []
+        for topic in self.topics:
+            knowledge_of_topics.append({topic.name: topic.knowledge})
+        return knowledge_of_topics
+
+    def summarize_by_topic(self):
+        for topic in self.topics:
+            topic.summarize(self.docs)

qlib/finco/llm.py

@@ -0,0 +1,111 @@
+import os
+import time
+import openai
+import json
+from typing import Optional
+from qlib.finco.conf import Config
+from qlib.finco.utils import SingletonBaseClass
+from qlib.finco.log import FinCoLog
+
+
+class APIBackend(SingletonBaseClass):
+    def __init__(self):
+        self.cfg = Config()
+        openai.api_key = self.cfg.openai_api_key
+        if self.cfg.use_azure:
+            openai.api_type = "azure"
+            openai.api_base = self.cfg.azure_api_base
+            openai.api_version = self.cfg.azure_api_version
+        self.use_azure = self.cfg.use_azure
+
+        self.debug_mode = False
+        if self.cfg.debug_mode:
+            self.debug_mode = True
+            cwd = os.getcwd()
+            self.cache_file_location = os.path.join(cwd, "prompt_cache.json")
+            self.cache = (
+                json.load(open(self.cache_file_location, "r")) if os.path.exists(self.cache_file_location) else {}
+            )
+
+    def build_messages_and_create_chat_completion(self, user_prompt, system_prompt=None, former_messages=[], **kwargs):
+        """build the messages to avoid implementing several redundant lines of code"""
+        cfg = Config()
+        # TODO: system prompt should always be provided. In development stage we can use default value
+        if system_prompt is None:
+            try:
+                system_prompt = cfg.system_prompt
+            except AttributeError:
+                FinCoLog().warning("system_prompt is not set, using default value.")
+                system_prompt = "You are an AI assistant who helps to answer user's questions about finance."
+        messages = [
+            {
+                "role": "system",
+                "content": system_prompt,
+            }
+        ]
+        messages.extend(former_messages[-1*cfg.max_past_message_include:])
+        messages.append(
+            {
+                "role": "user",
+                "content": user_prompt,
+            }
+        )
+        fcl = FinCoLog()
+        response = self.try_create_chat_completion(messages=messages, **kwargs)
+        fcl.log_message(messages)
+        fcl.log_response(response)
+        return response
+
+    def try_create_chat_completion(self, max_retry=10, **kwargs):
+        max_retry = self.cfg.max_retry if self.cfg.max_retry is not None else max_retry
+        for i in range(max_retry):
+            try:
+                response = self.create_chat_completion(**kwargs)
+                return response
+            except (openai.error.RateLimitError, openai.error.Timeout, openai.error.APIError) as e:
+                print(e)
+                print(f"Retrying {i+1}th time...")
+                time.sleep(1)
+                continue
+            except openai.InvalidRequestError as e:
+                print("Invalid request, will try to reduce the messages length and retry...")
+                if len(kwargs["messages"]) > 2:
+                    kwargs["messages"] = kwargs["messages"][[0]] + kwargs["messages"][3:]
+                    continue
+                raise e
+        raise Exception(f"Failed to create chat completion after {max_retry} retries.")
+
+    def create_chat_completion(
+        self,
+        messages,
+        model=None,
+        temperature: float = None,
+        max_tokens: Optional[int] = None,
+    ) -> str:
+
+        if self.debug_mode:
+            key = json.dumps(messages)
+            if key in self.cache:
+                return self.cache[key]
+
+        if temperature is None:
+            temperature = self.cfg.temperature
+        if max_tokens is None:
+            max_tokens = self.cfg.max_tokens
+
+        if self.cfg.use_azure:
+            response = openai.ChatCompletion.create(
+                engine=self.cfg.model,
+                messages=messages,
+                max_tokens=self.cfg.max_tokens,
+            )
+        else:
+            response = openai.ChatCompletion.create(
+                model=self.cfg.model,
+                messages=messages,
+            )
+        resp = response.choices[0].message["content"]
+        if self.debug_mode:
+            self.cache[key] = resp
+            json.dump(self.cache, open(self.cache_file_location, "w"))
+        return resp

qlib/finco/log.py

@@ -0,0 +1,131 @@
+"""
+This module will base on Qlib's logger module and provides some interactive functions.
+"""
+import logging
+
+from typing import Dict, List
+from qlib.finco.utils import SingletonBaseClass
+from contextlib import contextmanager
+
+
+class LogColors:
+    """
+    ANSI color codes for use in console output.
+    """
+    RED = "\033[91m"
+    GREEN = "\033[92m"
+    YELLOW = "\033[93m"
+    BLUE = "\033[94m"
+    MAGENTA = "\033[95m"
+    CYAN = "\033[96m"
+    WHITE = "\033[97m"
+    GRAY = "\033[90m"
+    BLACK = "\033[30m"
+
+    BOLD = "\033[1m"
+    ITALIC = "\033[3m"
+
+    END = "\033[0m"
+
+    @classmethod
+    def get_all_colors(cls):
+        names = dir(cls)
+        names = [name for name in names if not name.startswith("__") and not callable(getattr(cls, name))]
+        var_values = [getattr(cls, name) for name in names]
+        return var_values
+
+    def render(self, text: str, color: str = "", style: str = ""):
+        """
+        render text by input color and style. It's not recommend that input text is already rendered.
+        """
+        # This method is called too frequently, which is not good.
+        colors = self.get_all_colors()
+        # Perhaps color and font should be distinguished here.
+        if color:
+            assert color in colors, f"color should be in: {colors} but now is: {color}"
+        if style:
+            assert style in colors, f"style should be in: {colors} but now is: {style}"
+
+        text = f"{color}{text}{self.END}"
+        text = f"{style}{text}{self.END}"
+
+        return text
+
+
+@contextmanager
+def formatting_log(logger, title="Info"):
+    """
+    a context manager, print liens before and after a function
+    """
+    length = {"Start": 120, "Task": 120, "Info": 60, "Interact": 60, "End": 120}.get(title, 60)
+    color, bold = (LogColors.YELLOW, LogColors.BOLD) \
+        if title in ["Start", "Task", "Info", "Interact", "End"] else (LogColors.CYAN, "")
+    logger.info("")
+    logger.info(f"{color}{bold}{'-'} {title} {'-' * (length - len(title))}{LogColors.END}")
+    yield
+    logger.info("")
+
+
+class FinCoLog(SingletonBaseClass):
+    # TODO:
+    # - config to file logger and save it into workspace
+    def __init__(self) -> None:
+        self.logger = logging.Logger("interactive")
+        # TODO:  merge these with Qlib's default logger.
+        #  We can do the same thing by changing the default log dict of Qlib.
+        #  Reference: https://github.com/microsoft/qlib/blob/main/qlib/config.py#L155
+
+        handler = logging.StreamHandler()
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        self.logger.addHandler(handler)
+        self.logger.setLevel(logging.INFO)
+
+    def log_message(self, messages: List[Dict[str, str]]):
+        """
+        messages is some info like this  [
+            {
+                "role": "system",
+                "content": system_prompt,
+            },
+            {
+                "role": "user",
+                "content": user_prompt,
+            },
+        ]
+        """
+        with formatting_log(self.logger, "GPT Messages"):
+            for m in messages:
+                self.logger.info(
+                    f"{LogColors.MAGENTA}{LogColors.BOLD}Role:{LogColors.END} "
+                    f"{LogColors.CYAN}{m['role']}{LogColors.END}\n"
+                    + f"{LogColors.MAGENTA}{LogColors.BOLD}Content:{LogColors.END} "
+                      f"{LogColors.CYAN}{m['content']}{LogColors.END}\n")
+
+    def log_response(self, response: str):
+        with formatting_log(self.logger, "GPT Response"):
+            self.logger.info(
+                f"{LogColors.CYAN}{response}{LogColors.END}\n")
+
+    # TODO:
+    # It looks wierd if we only have logger
+    def info(self, *args, plain=False, title="Info"):
+        if plain:
+            return self.plain_info(*args)
+        with formatting_log(self.logger, title):
+            for arg in args:
+                self.logger.info(f"{LogColors.WHITE}{arg}{LogColors.END}")
+
+    def plain_info(self, *args):
+        for arg in args:
+            self.logger.info(
+                f"{LogColors.YELLOW}{LogColors.BOLD}Info:{LogColors.END}{LogColors.WHITE}{arg}{LogColors.END}")
+
+    def warning(self, *args):
+        for arg in args:
+            self.logger.warning(
+                f"{LogColors.BLUE}{LogColors.BOLD}Warning:{LogColors.END}{arg}")
+
+    def error(self, *args):
+        for arg in args:
+            self.logger.error(
+                f"{LogColors.RED}{LogColors.BOLD}Error:{LogColors.END}{arg}")

qlib/finco/prompt_template.py

@@ -0,0 +1,32 @@
+from typing import Union
+from pathlib import Path
+from jinja2 import Template
+import yaml
+
+from qlib.finco.utils import SingletonBaseClass
+from qlib.finco import get_finco_path
+
+
+class PromptTemplate(SingletonBaseClass):
+    def __init__(self) -> None:
+        super().__init__()
+        _template = yaml.load(open(Path.joinpath(get_finco_path(), "prompt_template.yaml"), "r"),
+                              Loader=yaml.FullLoader)
+        for k, v in _template.items():
+            if k == "mods":
+                continue
+            self.__setattr__(k, Template(v))
+
+    def get(self, key: str):
+        return self.__dict__.get(key, Template(""))
+
+    def update(self, key: str, value):
+        self.__setattr__(key, value)
+
+    def save(self, file_path: Union[str, Path]):
+        if isinstance(file_path, str):
+            file_path = Path(file_path)
+        Path.mkdir(file_path.parent, exist_ok=True)
+
+        with open(file_path, 'w') as f:
+            yaml.dump(self.__dict__, f)

qlib/finco/tpl/README.md

@@ -0,0 +1,12 @@
+This is a set of templates that should be copied for a new project.
+
+Here are the explanations for the templates folder.
+
+| folder | explanations                                                     |
+|--------|------------------------------------------------------------------|
+| sl     | Default configuration for supervised learning                    |
+| sl-cfg | Like configuration in sl. But the dataset is highly configurable |
+
+
+# TODO
+- [ ] [Copier](https://copier.readthedocs.io/en/stable/#quick-start) may be useful if the generation process becomes complicated

qlib/finco/tpl/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+from pathlib import Path
+
+DIRNAME = Path(__file__).absolute().resolve().parent
+
+
+def get_tpl_path() -> Path:
+    """
+    return the template path
+    Because the template path is located in the folder. We don't know where it is located. So __file__ for this module will be used.
+    """
+    return DIRNAME

qlib/finco/tpl/sl/workflow_config.yaml

@@ -0,0 +1,73 @@
+qlib_init:
+    provider_uri: "~/.qlib/qlib_data/cn_data"
+    region: cn
+experiment_name: finCo
+market: &market csi300
+benchmark: &benchmark SH000300
+data_handler_config: &data_handler_config
+    start_time: 2008-01-01
+    end_time: 2020-08-01
+    fit_start_time: 2008-01-01
+    fit_end_time: 2014-12-31
+    instruments: *market
+port_analysis_config: &port_analysis_config
+    strategy:
+        class: TopkDropoutStrategy
+        module_path: qlib.contrib.strategy
+        kwargs:
+            model: <MODEL>
+            dataset: <DATASET>
+            topk: 50
+            n_drop: 5
+    backtest:
+        start_time: 2017-01-01
+        end_time: 2020-08-01
+        account: 100000000
+        benchmark: *benchmark
+        exchange_kwargs:
+            limit_threshold: 0.095
+            deal_price: close
+            open_cost: 0.0005
+            close_cost: 0.0015
+            min_cost: 5
+task:
+    model:
+        class: LGBModel
+        module_path: qlib.contrib.model.gbdt
+        kwargs:
+            loss: mse
+            colsample_bytree: 0.8879
+            learning_rate: 0.2
+            subsample: 0.8789
+            lambda_l1: 205.6999
+            lambda_l2: 580.9768
+            max_depth: 8
+            num_leaves: 210
+            num_threads: 20
+    dataset:
+        class: DatasetH
+        module_path: qlib.data.dataset
+        kwargs:
+            handler:
+                class: Alpha158
+                module_path: qlib.contrib.data.handler
+                kwargs: *data_handler_config
+            segments:
+                train: [2008-01-01, 2014-12-31]
+                valid: [2015-01-01, 2016-12-31]
+                test: [2017-01-01, 2020-08-01]
+    record: 
+        - class: SignalRecord
+          module_path: qlib.workflow.record_temp
+          kwargs: 
+            model: <MODEL>
+            dataset: <DATASET>
+        - class: SigAnaRecord
+          module_path: qlib.workflow.record_temp
+          kwargs: 
+            ana_long_short: False
+            ann_scaler: 252
+        - class: PortAnaRecord
+          module_path: qlib.workflow.record_temp
+          kwargs: 
+            config: *port_analysis_config

qlib/finco/utils.py

@@ -0,0 +1,38 @@
+import json
+
+from fuzzywuzzy import fuzz
+
+
+class SingletonMeta(type):
+    _instance = None
+
+    def __call__(cls, *args, **kwargs):
+        if cls._instance is None:
+            cls._instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
+        return cls._instance
+
+
+class SingletonBaseClass(metaclass=SingletonMeta):
+    """
+    Because we try to support defining Singleton with `class A(SingletonBaseClass)` instead of `A(metaclass=SingletonMeta)`
+    This class becomes necessary
+
+    """
+    # TODO: Add move this class to Qlib's general utils.
+
+
+def parse_json(response):
+    try:
+        return json.loads(response)
+    except json.decoder.JSONDecodeError:
+        pass
+
+    raise Exception(f"Failed to parse response: {response}, please report it or help us to fix it.")
+
+
+def similarity(text1, text2):
+    text1 = text1 if isinstance(text1, str) else ""
+    text2 = text2 if isinstance(text2, str) else ""
+
+    # Maybe we can use other similarity algorithm such as tfidf
+    return fuzz.ratio(text1, text2)

qlib/finco/workflow.py

@@ -0,0 +1,223 @@
+import sys
+import copy
+import shutil
+from pathlib import Path
+from typing import List
+
+from qlib.finco.task import HighLevelPlanTask, SummarizeTask, TrainTask
+from qlib.finco.prompt_template import PromptTemplate, Template
+from qlib.finco.log import FinCoLog, LogColors
+from qlib.finco.utils import similarity
+from qlib.finco.llm import APIBackend
+from qlib.finco.conf import Config
+from qlib.finco.knowledge import KnowledgeBase, Topic
+
+
+class WorkflowContextManager:
+    """Context Manager stores the context of the workflow"""
+
+    """All context are key value pairs which saves the input, output and status of the whole workflow"""
+
+    def __init__(self) -> None:
+        self.context = {}
+        self.logger = FinCoLog()
+
+    def set_context(self, key, value):
+        if key in self.context:
+            self.logger.warning("The key already exists in the context, the value will be overwritten")
+        self.context[key] = value
+
+    def get_context(self, key):
+        # NOTE: if the key doesn't exist, return None. In the future, we may raise an error to detect abnormal behavior
+        if key not in self.context:
+            self.logger.warning("The key doesn't exist in the context")
+            return None
+        return self.context[key]
+
+    def update_context(self, key, new_value):
+        # NOTE: if the key doesn't exist, return None. In the future, we may raise an error to detect abnormal behavior
+        if key not in self.context:
+            self.logger.warning("The key doesn't exist in the context")
+        self.context.update({key: new_value})
+
+    def get_all_context(self):
+        """return a deep copy of the context"""
+        """TODO: do we need to return a deep copy?"""
+        return copy.deepcopy(self.context)
+
+    def retrieve(self, query: str) -> dict:
+        if query in self.context.keys():
+            return {query: self.context.get(query)}
+
+        # Note: retrieve information from context by string similarity maybe abandon in future
+        scores = {}
+        for k, v in self.context.items():
+            scores.update({k: max(similarity(query, k), similarity(query, v))})
+        max_score_key = max(scores, key=scores.get)
+        return {max_score_key: self.context.get(max_score_key)}
+
+    def clear(self, reserve: list = None):
+        if reserve is None:
+            reserve = []
+
+        _context = {k: self.get_context(k) for k in reserve}
+        self.context = _context
+
+
+class WorkflowManager:
+    """This manage the whole task automation workflow including tasks and actions"""
+
+    def __init__(self, workspace=None) -> None:
+        self.logger = FinCoLog()
+
+        if workspace is None:
+            self._workspace = Path.cwd() / "finco_workspace"
+        else:
+            self._workspace = Path(workspace)
+        self.conf = Config()
+        self._confirm_and_rm()
+
+        self.prompt_template = PromptTemplate()
+        self.context = WorkflowContextManager()
+        self.context.set_context("workspace", self._workspace)
+        self.default_user_prompt = "Please help me build a low turnover strategy that focus more on longterm return in China A csi300. Please help to use lightgbm model."
+
+    def _confirm_and_rm(self):
+        # if workspace exists, please confirm and remove it. Otherwise exit.
+        if self._workspace.exists() and not self.conf.continuous_mode:
+            self.logger.info(title="Interact")
+            flag = input(
+                LogColors().render(
+                    f"Will be deleted: \n\t{self._workspace}\n"
+                    f"If you do not need to delete {self._workspace},"
+                    f" please change the workspace dir or rename existing files\n"
+                    f"Are you sure you want to delete, yes(Y/y), no (N/n):",
+                    color=LogColors.WHITE)
+            )
+            if str(flag) not in ["Y", "y"]:
+                sys.exit()
+            else:
+                # remove self._workspace
+                shutil.rmtree(self._workspace)
+        elif self._workspace.exists() and self.conf.continuous_mode:
+            shutil.rmtree(self._workspace)
+
+    def set_context(self, key, value):
+        """Direct call set_context method of the context manager"""
+        self.context.set_context(key, value)
+
+    def get_context(self) -> WorkflowContextManager:
+        return self.context
+
+    def run(self, prompt: str) -> Path:
+        """
+        The workflow manager is supposed to generate a codebase based on the prompt
+
+        Parameters
+        ----------
+        prompt: str
+            the prompt user gives
+
+        Returns
+        -------
+        Path
+            The workflow manager is expected to produce output that includes a codebase containing generated code, results, and reports in a designated location.
+            The path is returned
+
+            The output path should follow a specific format:
+            - TODO: design
+              There is a summarized report where user can start from.
+        """
+
+        # NOTE: The following items are not designed to make the workflow very flexible.
+        # - The generated tasks can't be changed after geting new information from the execution retuls.
+        #   - But it is required in some cases, if we want to build a external dataset, it maybe have to plan like autogpt...
+
+        # NOTE: default user prompt might be changed in the future and exposed to the user
+        if prompt is None:
+            self.set_context("user_prompt", self.default_user_prompt)
+        else:
+            self.set_context("user_prompt", prompt)
+        self.logger.info(f"user_prompt: {self.get_context().get_context('user_prompt')}", title="Start")
+
+        # NOTE: list may not be enough for general task list
+        task_list = [HighLevelPlanTask(), SummarizeTask()]
+        task_finished = []
+        while len(task_list):
+            task_list_info = [str(task) for task in task_list]
+
+            # task list is not long, so sort it is not a big problem
+            # TODO: sort the task list based on the priority of the task
+            # task_list = sorted(task_list, key=lambda x: x.task_type)
+            t = task_list.pop(0)
+            self.logger.info(f"Task finished: {[str(task) for task in task_finished]}",
+                             f"Task in queue: {task_list_info}",
+                             f"Executing task: {str(t)}",
+                             title="Task")
+
+            t.assign_context_manager(self.context)
+            res = t.execute()
+            t.summarize()
+            task_finished.append(t)
+            self.context.set_context("task_finished", task_finished)
+            self.logger.plain_info(f"{str(t)} finished.\n\n\n")
+
+            task_list = res + task_list
+
+        return self._workspace
+
+
+class LearnManager:
+    __DEFAULT_TOPICS = ["IC", "MaxDropDown"]
+
+    def __init__(self):
+        self.epoch = 0
+        self.wm = WorkflowManager()
+
+        topics = [Topic(name=topic, describe=self.wm.prompt_template.get(f"Topic_{topic}")) for topic in
+                  self.__DEFAULT_TOPICS]
+        self.knowledge_base = KnowledgeBase(init_path=Path.cwd().joinpath('knowledge'), topics=topics)
+
+    def run(self, prompt):
+        # todo: add early stop condition
+        for i in range(10):
+            self.wm.run(prompt)
+            self.knowledge_base.update(self.wm._workspace)
+            self.knowledge_base.summarize_by_topic()
+            self.learn()
+            self.epoch += 1
+
+    def learn(self):
+        workspace = self.wm.context.get_context("workspace")
+
+        def _drop_duplicate_task(_task: List):
+            unique_task = {}
+            for obj in _task:
+                task_name = obj.__class__.__name__
+                if task_name not in unique_task:
+                    unique_task[task_name] = obj
+            return list(unique_task.values())
+
+        # one task maybe run several times in workflow
+        task_finished = _drop_duplicate_task(self.wm.context.get_context("task_finished"))
+
+        user_prompt = self.wm.context.get_context("user_prompt")
+        summary = self.wm.context.get_context("summary")
+
+        for task in task_finished:
+            prompt_workflow_selection = self.wm.prompt_template.get(f"{self.__class__.__name__}_user").render(
+                summary=summary, brief=self.knowledge_base.query_topics(),
+                task_finished=[str(t) for t in task_finished],
+                task=task.__class__.__name__, system=task.system.render(), user_prompt=user_prompt
+            )
+
+            response = APIBackend().build_messages_and_create_chat_completion(
+                user_prompt=prompt_workflow_selection,
+                system_prompt=self.wm.prompt_template.get(f"{self.__class__.__name__}_system").render()
+            )
+
+            # todo: response assertion
+            task.prompt_template.update(key=f"{task.__class__.__name__}_system", value=Template(response))
+
+        self.wm.prompt_template.save(Path.joinpath(workspace, f"prompts/checkpoint_{self.epoch}.yml"))
+        self.wm.context.clear(reserve=["workspace"])

qlib/workflow/record_temp.py

@@ -18,7 +18,7 @@ from ..utils import fill_placeholder, flatten_dict, class_casting, get_date_by_s
 from ..utils.time import Freq
 from ..utils.data import deepcopy_basic_type
 from ..contrib.eva.alpha import calc_ic, calc_long_short_return, calc_long_short_prec
-
+from qlib.contrib.analyzer import HFAnalyzer, SignalAnalyzer
 
 logger = get_module_logger("workflow", logging.INFO)
 
@@ -156,6 +156,9 @@ class RecordTemp:
                 with class_casting(self, self.depend_cls):
                     self.check(include_self=True)
 
+    def analyse(self):
+        raise NotImplementedError(f"Please implement the `analysis` method.")
+
 
 class SignalRecord(RecordTemp):
     """

scripts/finco/README.md

@@ -0,0 +1,15 @@
+
+
+# Requirements
+
+
+Use following install command to complete the project.
+```
+pip install -e '.[finco]'
+```
+
+
+# TODOs
+
+- [ ] Select the appropriate LLM API
+  - Which API is more suitable for meeting our requirements - the original API or an alternative like LangChain?

scripts/finco/cmd.sh

@@ -0,0 +1,15 @@
+#!/bin/bash
+set -x # show command
+set -e # Error on exception
+
+DIR="$(
+	cd "$(dirname "$(readlink -f "$0")")" || exit
+	pwd -P
+)"
+# --load the cridentials
+if [ -e $DIR/cridential.sh ]; then
+	source $DIR/cridential.sh
+fi
+
+# run the command
+python -m qlib.finco.cli "please help me build a low turnover strategy that focus more on longterm return"

scripts/finco/cridential.sh.example

@@ -0,0 +1,3 @@
+export OPENAI_API_TYPE=azure  # This only necessary for Azure OpenAI
+export OPENAI_API_KEY=
+export OPENAI_API_BASE=

setup.py

@@ -173,6 +173,14 @@ setup(
             "tianshou<=0.4.10",
             "torch",
         ],
+        "finco": [
+            # finco is not necessary for all Qlib users; So a single require section is used for it.
+            "openapi",
+            "pydantic",  # Please add it to basic requirements after the design of pydantic is state.
+            "python-dotenv",  # I don't think this is necessary if we use pydantic.
+            "fuzzywuzzy",
+            "python-Levenshtein"    # not necessary but accelerate fuzzywuzzy calculation
+        ],
     },
     include_package_data=True,
     classifiers=[

tests/finco/test_cfg.py

@@ -0,0 +1,71 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+import unittest
+import shutil
+import difflib
+from qlib.finco.tpl import get_tpl_path
+import ruamel.yaml as yaml
+
+from qlib.data.dataset.handler import DataHandlerLP
+from qlib.utils import init_instance_by_config
+from qlib.tests import TestAutoData
+
+from pathlib import Path
+from qlib.finco.tpl import get_tpl_path
+from qlib.finco.task import YamlEditTask
+
+DIRNAME = Path(__file__).absolute().resolve().parent
+
+
+class FincoTpl(TestAutoData):
+    def test_tpl_consistence(self):
+        """Motivation: make sure the configuable template is consistent with the default config"""
+        tpl_p = get_tpl_path()
+        with (tpl_p / "sl" / "workflow_config.yaml").open("rb") as fp:
+            config = yaml.safe_load(fp)
+        # init_data_handler
+        hd: DataHandlerLP = init_instance_by_config(config["task"]["dataset"]["kwargs"]["handler"])
+        # NOTE: The config in workflow_config.yaml is generated by the following code:
+        # dump in yaml format to file without auto linebreak
+        # print(yaml.dump(hd.data_loader.fields, width=10000, stream=open("_tmp", "w")))
+
+        with (tpl_p / "sl-cfg" / "workflow_config.yaml").open("rb") as fp:
+            config = yaml.safe_load(fp)
+        hd_ds: DataHandlerLP = init_instance_by_config(config["task"]["dataset"]["kwargs"]["handler"])
+        self.assertEqual(hd_ds.data_loader.fields, hd.data_loader.fields)
+
+        check = hd_ds.fetch().fillna(0.0) == hd.fetch().fillna(0.0)
+        self.assertTrue(check.all().all())
+
+    def test_update_yaml(self):
+        p = get_tpl_path() / "sl" / "workflow_config.yaml"
+        p_new = DIRNAME / "_test_config.yaml"
+        shutil.copy(p, p_new)
+        updated_content = """
+class: LGBModelTest
+module_path: qlib.contrib.model.gbdt
+kwargs:
+    loss: mse
+    colsample_bytree: 1.8879
+    learning_rate: 0.3
+    subsample: 0.8790
+    lambda_l1: 205.7000
+    lambda_l2: 580.9769
+    max_depth: 9
+    num_leaves: 211
+    num_threads: 21
+"""
+        t = YamlEditTask(p_new, "task.model", updated_content)
+        t.execute()
+        # NOTE: the formmat is changed by ruamel.yaml, so it can't be compared by text directly..
+        # print the diff between p and p_new with difflib
+        # with p.open("r") as fp:
+        #     content = fp.read()
+        # with p_new.open("r") as fp:
+        #     content_new = fp.read()
+        # for line in difflib.unified_diff(content, content_new, fromfile="original", tofile="new", lineterm=""):
+        #     print(line)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/finco/test_sumarize.py

@@ -0,0 +1,66 @@
+import unittest
+import os
+import shutil
+
+from dotenv import load_dotenv
+# pydantic support load_dotenv,   so load_dotenv will be deprecated in the future.
+
+from qlib.finco.task import SummarizeTask
+from qlib.finco.workflow import WorkflowContextManager
+from qlib.finco.llm import APIBackend
+from qlib.finco.workflow import WorkflowManager
+
+load_dotenv(verbose=True, override=True)
+
+
+class TestSummarize(unittest.TestCase):
+
+    def test_chat(self):
+        messages = [
+            {
+                "role": "system",
+                "content": "Your are a professional financial assistant.",
+            },
+            {
+                "role": "user",
+                "content": "How to write a perfect quant strategy.",
+            },
+        ]
+        response = APIBackend().try_create_chat_completion(messages=messages)
+        print(response)
+
+    def test_execution(self):
+        task = SummarizeTask()
+        context = WorkflowContextManager()
+        context.set_context("workspace", "../../examples/benchmarks/Linear")
+        context.set_context("user_prompt", "My main focus is on the performance of the strategy's return."
+                                           "Please summarize the information and give me some advice.")
+        task.assign_context_manager(context)
+        resp = task.execute()
+        print(resp)
+
+    def test_generate_batch_result(self):
+        wm = WorkflowManager()
+
+        prompt = wm.default_user_prompt
+        # prompt = ""
+
+        workdir = os.path.dirname(wm.get_context().get_context("workspace"))
+        summaries_path = os.path.join(workdir, "summaries")
+
+        if not os.path.exists(summaries_path):
+            os.makedirs(summaries_path)
+
+        for i in range(10):
+            wm.run(prompt)
+            if os.path.exists(f"{workdir}/finCoReport.md"):
+                shutil.move(f"{workdir}/finCoReport.md", f"{workdir}/summaries/finCoReport{i}.md")
+
+    def test_parse2txt(self):
+        task = SummarizeTask()
+        resp = task.get_info_from_file("")
+        print(resp)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/finco/test_utils.py

@@ -0,0 +1,23 @@
+import unittest
+from qlib.finco.utils import SingletonBaseClass
+
+
+class TimeUtils(unittest.TestCase):
+
+    def test_singleton(self):
+        # self.assertEqual(self.to_str(data.tail()), self.to_str(res))
+        closure_checker = []
+
+        class A(SingletonBaseClass):
+
+            def __init__(self) -> None:
+                closure_checker.append(0)
+
+        A()
+        self.assertEqual(len(closure_checker), 1)
+        A()
+        self.assertEqual(len(closure_checker), 1)
+
+
+if __name__ == "__main__":
+    unittest.main()