"""Access to account statement webservice."""import loggingimport osimport timeimport xml.etree.ElementTree as etfrom contextlib importsuppressfrom typing importFinalfrom urllib.parse importurlparsefrom urllib.request importurlopenfrom ib\_async importutilfrom ib\_async.objects importDynamicObject\_logger=logging.getLogger("ib\_async.flexreport")FLEXREPORT\_URL:Final=("https://ndcdyn.interactivebrokers.com/AccountManagement/""FlexWebService/SendRequest?")"""https://www.interactivebrokers.com/campus/ibkr-api-page/flex-web-service/#flex-generate-report"""
[[docs]](../../api.html#ib_async.flexreport.FlexError)class FlexError(Exception):pass

[[docs]](../../api.html#ib_async.flexreport.FlexReport)class FlexReport:""" To obtain a token: \* Login to web portal \* Go to Settings \* Click on "Configure Flex Web Service" \* Generate token """data: bytesroot: et.Elementdef \_\_init\_\_(self, token=None, queryId=None, path=None):""" Download a report by giving a valid ``token`` and ``queryId``, or load from file by giving a valid ``path``. To overwrite default URL, set env variable ``IB\_FLEXREPORT\_URL``. """if token and queryId:self.download(token, queryId)elif path:self.load(path)
[[docs]](../../api.html#ib_async.flexreport.FlexReport.topics)def topics(self):"""Get the set of topics that can be extracted from this report."""return set(node.tag for node in self.root.iter() if node.attrib)

[[docs]](../../api.html#ib_async.flexreport.FlexReport.extract)def extract(self, topic: str, parseNumbers=True) -\> list:""" Extract items of given topic and return as list of objects. The topic is a string like TradeConfirm, ChangeInDividendAccrual, Order, etc. """cls = type(topic, (DynamicObject,), {})results = [cls(\*\*node.attrib) for node in self.root.iter(topic)]if parseNumbers:for obj in results:d = obj.\_\_dict\_\_for k, v in d.items():with suppress(ValueError):d[k] = float(v)d[k] = int(v)return results

[[docs]](../../api.html#ib_async.flexreport.FlexReport.df)def df(self, topic: str, parseNumbers=True):"""Same as extract but return the result as a pandas DataFrame."""return util.df(self.extract(topic, parseNumbers))

[[docs]](../../api.html#ib_async.flexreport.FlexReport.get_url)def get\_url(self):"""Generate flexreport URL."""def is\_valid\_url(url: str) -\> bool:try:result = urlparse(url)# Must have scheme (http/https) and netloc (domain)return all([result.scheme, result.netloc])except Exception:return False\_url = os.getenv("IB\_FLEXREPORT\_URL", FLEXREPORT\_URL)if is\_valid\_url(\_url):return \_urlraise FlexError("Invalid URL, please check that env variable IB\_FLEXREPORT\_URL is set correctly.")

[[docs]](../../api.html#ib_async.flexreport.FlexReport.download)def download(self, token, queryId):"""Download report for the given ``token`` and ``queryId``."""base\_url = self.get\_url()query = f"t={token}&q={queryId}&v=3"url = base\_url + queryresp = urlopen(url)data = resp.read()root = et.fromstring(data)elem = root.find("Status")if elem is not None and elem.text == "Success":elem = root.find("ReferenceCode")assert elem is not Nonecode = elem.textelem = root.find("Url")assert elem is not NonebaseUrl = elem.text\_logger.info("Statement is being prepared...")else:elem = root.find("ErrorCode")errorCode = elem.text if elem is not None else ""elem = root.find("ErrorMessage")errorMsg = elem.text if elem is not None else ""raise FlexError(f"{errorCode}: {errorMsg}")while True:time.sleep(1)url = f"{baseUrl}?q={code}&t={token}&v=3"resp = urlopen(url)self.data = resp.read()self.root = et.fromstring(self.data)if self.root[0].tag == "code":msg = self.root[0].textif msg and msg.startswith("Statement generation in progress"):\_logger.info("still working...")continueelse:raise FlexError(msg)break\_logger.info("Statement retrieved.")

[[docs]](../../api.html#ib_async.flexreport.FlexReport.load)def load(self, path):"""Load report from XML file."""with open(path, "rb") as f:self.data = f.read()self.root = et.fromstring(self.data)

[[docs]](../../api.html#ib_async.flexreport.FlexReport.save)def save(self, path):"""Save report to XML file."""with open(path, "wb") as f:f.write(self.data)

if\_\_name\_\_=="\_\_main\_\_":util.logToConsole()report=FlexReport("945692423458902392892687","272555")print(report.topics())trades=report.extract("Trade")print(trades)