import json import time import requests import pandas as pd from DBHelper.MongoHelper import MongoHelper from RatingData.scripts.risk_detail import associate_risk_detail, change_log_detail from Utils.ErrorUtil import LogConditionCheckFailed from CompanyData.CompanyDataObj import CompanyData, BasicInfo, ShareHolder, MainMember, ComprehensiveCreditAnalysis, \ FinancialElementsAnalysis, BusinessElementsAnalysis, FinancialIndex, BalanceSheet, ProfitSheet, AppendixDataSheet, \ EsgRatingAnalysis, FinancialIndexs from Utils.CommonUtil import sub_dict, df_iterrows class CompanyDataImpl(CompanyData): """企业数据实现类""" db_tyc = MongoHelper("tyc") db_tfse = MongoHelper("tfse_v0.21") def drag_data_from_tyc_api(self): """拉取天眼查接口数据""" url = "http://api.fecribd.com/api/tyc/drag_data" headers = {'token': "uzdq51N4!I0%HY4sCaQ!aeCSIDIVIdAM"} data = {"企业名称": self.name} res = requests.post(url=url, headers=headers, data=json.dumps(data)) return True if res.status_code == 200 else False def prepare_company_init_data(self): def prepare_business_data(): data = self.db_tyc.find_single_column( "公司背景", "基本信息", {"企业名称": self.name}, "基本信息" ) basic_info = BasicInfo() basic_info.status = data['regStatus'] basic_info.legal_person = data['legalPersonName'] basic_info.company_type = data['companyOrgType'] basic_info.taxpayer_id = data['taxNumber'] basic_info.business_scope = data['businessScope'] basic_info.registered_capital = data['regCapital'] basic_info.paid_capital = data['actualCapital'] basic_info.registered_address = data['regLocation'] basic_info.registration_authority = data['regInstitute'] basic_info.industry = data['industry'] basic_info.staff_size = data['staffNumRange'] basic_info.people_insured_num = data['socialStaffNum'] basic_info.micro_company = "是" if data['regStatus'] == 1 else "否" self.basic_info = basic_info def prepare_share_holders(): try: self.share_holders = list() data = self.db_tyc.find_single_column( "公司背景", "企业股东", {"企业名称": self.name}, "企业股东" ) try: results = data['result'] except KeyError: raise LogConditionCheckFailed("企业股东数据异常", 200) for result in results: share_holder = ShareHolder() share_holder.name = result['name'] share_holder.share_holder_type = "公司" if result['type'] == 1 else ( "个人" if result['type'] == 2 else "其他") if result.__contains__('capital'): share_holder.share_holding_ratio = None if result['capital'][0]['percent'] == '' else \ result['capital'][0]['percent'] share_holder.subscription_amount = None if result['capital'][0]['amomon'] == '' else \ result['capital'][0]['amomon'] share_holder.subscription_date = None if result['capital'][0]['time'] == '' else \ result['capital'][0]['time'] else: share_holder.share_holding_ratio = None share_holder.subscription_amount = None share_holder.subscription_date = None if result.__contains__('capitalActl'): share_holder.paid_amount = [capital_actl['amomon'] for capital_actl in result['capitalActl']] share_holder.payment_method = [capital_actl['paymet'] for capital_actl in result['capitalActl']] share_holder.payment_time = [capital_actl['time'] for capital_actl in result['capitalActl']] else: share_holder.paid_amount = None share_holder.payment_method = None share_holder.payment_time = None self.share_holders.append(share_holder) except TypeError: self.share_holders = list() data = self.db_tyc.find_single_column( "公司背景", "十大股东", {"企业名称": self.name}, "十大股东" ) try: results = data['holderList'] except KeyError: raise LogConditionCheckFailed("企业股东数据异常", 200) for result in results: share_holder = ShareHolder() share_holder.name = result['name'] share_holder.share_holder_type = "公司" if result['cType'] == 1 else ( "个人" if result['cType'] == 2 else "其他") share_holder.share_holding_ratio = result['proportion'] share_holder.subscription_amount = None share_holder.subscription_date = None share_holder.paid_amount = None share_holder.payment_method = None share_holder.payment_time = None self.share_holders.append(share_holder) def prepare_main_members(): self.main_members = list() data = self.db_tyc.find_single_column( "公司背景", "主要人员", {"企业名称": self.name}, "主要人员" ) try: results = data['result'] except KeyError: raise LogConditionCheckFailed("主要人员数据异常", 200) for result in results: main_member = MainMember() main_member.name = result['name'] main_member.job_title = result['typeJoin'] self.main_members.append(main_member) def prepare_industry_l1(): """一级行业数据""" db = MongoHelper("tfse_v0.21") tyc_industry_l2 = self.basic_info.dict_keys_toggle()['行业'] fecr_industry_l1 = db.find_single_column( "模型端", "天眼查公司行业分类", {"二级行业": tyc_industry_l2}, "远东_一级行业" ) self.industry_l1 = fecr_industry_l1 def save_to_basic_business_info(): self.db_tfse.upsert_single_data( "企业数据", "基本工商信息", {"企业ID": self.cid}, { "企业ID": self.cid, "企业名称": self.name, "更新日期": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "工商信息": self.basic_info.dict_keys_toggle(), "股东信息": [item.dict_keys_toggle() for item in self.share_holders], "主要成员": [item.dict_keys_toggle() for item in self.main_members] } ) def save_to_update_summary(): self.update_time = { "工商信息": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()), "财务信息": None, "综信评价": None, "ESG评价": None, } self.db_tfse.upsert_single_data( "企业数据", "企业数据_更新汇总", {"企业ID": self.cid}, self.dict_keys_toggle() ) def __main__(): prepare_business_data() prepare_share_holders() prepare_main_members() prepare_industry_l1() save_to_basic_business_info() save_to_update_summary() __main__() def update_credit_data(self, rid): """更新综信评价相关数据""" def update_rating_result(): """更新评价结果""" rating_result = self.db_tfse.find_single_data( '综信评价数据', '评价结果', {'评价ID': rid}, ['企业ID', '信用等级', '信用评分', '评价年度', '经营评分', '财务评分', '风险评分', '评价时间'] ) self.cid = rating_result['企业ID'] del rating_result['企业ID'] rating_result['评价ID'] = rid self.cc_rating_result = rating_result def financial_indicator(): """更新财务指标""" financial_result = self.db_tfse.find_single_data( '综信评价数据', '财指结果', {'企业ID': self.cid}, ['净资产收益率', '存货周转率', '已获利息倍数', '应收账款周转率', '总资产周转率', '总资产增长率', '总资产报酬率', '技术投入比率', '营业增长率', '资产负债率', '速动比率'] ) financial_index = FinancialIndex() financial_index.roe = financial_result['净资产收益率'] financial_index.inventory_turnover = financial_result['存货周转率'] financial_index.interest_multiple = financial_result['已获利息倍数'] financial_index.accounts_receivable_turnover = financial_result['应收账款周转率'] financial_index.total_asset_turnover = financial_result['总资产周转率'] financial_index.total_asset_growth_rate = financial_result['总资产增长率'] financial_index.roa = financial_result['总资产报酬率'] financial_index.technology_investment_ratio = financial_result['技术投入比率'] financial_index.operating_growth_rate = financial_result['营业增长率'] financial_index.assets_and_liabilities = financial_result['资产负债率'] financial_index.quick_ratio = financial_result['速动比率'] self.fin_index = financial_index def financial_sheet(): """更新财务表""" fill_data = self.db_tfse.find_single_data( '综信评价数据', '填报数据', {'评价ID': rid}, ['财务填报'] ) def update_balance_sheet(): """资产负债表""" balance_data = fill_data['财务填报']['资产负债表'][0] balance = BalanceSheet() balance.accounts_receivable = balance_data['应收账款'] balance.stock = balance_data['存货'] balance.total_current_assets = balance_data['流动资产合计'] balance.total_assets = balance_data['资产总计'] balance.short_loan = balance_data['短期借款'] balance.one_year_liabilities = balance_data['一年内到期非流动负债'] balance.total_current_liabilities = balance_data['流动负债合计'] balance.long_term_loan = balance_data['长期借款'] balance.total_liabilities = balance_data['负债合计'] balance.total_owners_equity = balance_data['所有者权益合计'] self.balance_sheet = balance def update_profit_sheet(): """利润表""" profit_data = fill_data['财务填报']['利润表'][0] profit = ProfitSheet() profit.operating_income = profit_data['营业收入'] profit.operating_cost = profit_data['营业成本'] profit.total_profit = profit_data['利润总额'] profit.net_profit = profit_data['净利润'] self.profit_sheet = profit def update_appendix_sheet(): """补充数据表""" appendix_data = fill_data['财务填报']['补充数据表'][0] appendix = AppendixDataSheet() appendix.rd_expenses = appendix_data['研发费用'] appendix.interest_disbursement = appendix_data['计入财务费的利息支出'] appendix.interest_expense = appendix_data['资本化利息支出'] self.appendix_sheet = appendix def _main_(): update_balance_sheet() update_profit_sheet() update_appendix_sheet() _main_() def __main__(): update_rating_result() financial_indicator() financial_sheet() result = self.dict_to_save() del result['企业ID'] self.db_tfse.update_single_data( '企业数据', '企业数据_更新汇总', {'企业ID': self.cid}, result ) __main__() def update_esg_data(self, rid): """更新ESG相关数据""" rating_result = self.db_tfse.find_single_data( 'ESG评价数据', '评价结果', {'评价ID': rid}, ['企业ID', '评价等级', '评价年度', 'ESG得分', '环境得分', '社会得分', '公司治理得分', '评价时间'] ) cid = rating_result['企业ID'] del rating_result['企业ID'] rating_result['评价ID'] = rid self.esg_rating_result = rating_result result = self.dict_to_save() self.db_tfse.update_single_data( '企业数据', '企业数据_更新汇总', {'企业ID': cid}, result ) class CreditAnalysisImpl(ComprehensiveCreditAnalysis): """综合信用分析类""" db = MongoHelper("tfse_v0.21") def cc_rating_etl(self): rating_record = self.db.find_single_data( '企业数据', '评价记录', {"评价ID": self.rid}, ["报告fid"] ) rating_result = self.db.find_single_data( '综信评价数据', '评价结果', {"评价ID": self.rid}, ["企业ID", "企业名称", "经营评分", "财务评分", "风险评分", "评价年度"] ) rating_results = self.db.find_all_data( '综信评价数据', '评价结果', {"企业ID": rating_result['企业ID']}, ['信用等级', "信用评分", '评价时间', "评价方式"] ) text_model = self.db.find_single_data( '综信评价数据', '报告数据', {"评价ID": self.rid}, ["行业选择", "报告内容"] ) df_records = pd.DataFrame(rating_results).sort_values('评价时间', ascending=False) def result_head(): """结果数据""" self.cid = rating_result['企业ID'] self.name = rating_result['企业名称'] self.update_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) def make_radar_chart(): credit_performence = self.CreditPerformance() indicator_radar = credit_performence.IndicatorRadar() """综合信用指标表现雷达图数据""" max_score = indicator_radar.MaxScore() max_score.environment = 10 max_score.social_responsibility = 10 max_score.corporate_governance = 16 max_score.profitability = 18 max_score.asset_quality = 18 max_score.debt_risk = 18 max_score.operational_growth = 18 max_score.compliance_risk = 43 max_score.cperational_risk = 10 max_score.associated_risk = 10 index_score = indicator_radar.IndexScore() index_score.environment = rating_result['经营评分']["环境"] index_score.social_responsibility = rating_result['经营评分']["社会责任"] index_score.corporate_governance = rating_result['经营评分']["公司治理"] index_score.profitability = rating_result['财务评分']["盈利能力"]['合计'] index_score.asset_quality = rating_result['财务评分']["资产质量"]['合计'] index_score.debt_risk = rating_result['财务评分']["债务风险"]['合计'] index_score.operational_growth = rating_result['财务评分']["经营增长"]['合计'] index_score.compliance_risk = round(43 - rating_result['风险评分']["合规风险"], 2) index_score.cperational_risk = round(10 - rating_result['风险评分']["经营风险"]['合计'], 2) index_score.associated_risk = round(10 - rating_result['风险评分']["关联风险"]['合计'], 2) indicator_radar.index_score = index_score indicator_radar.max_score = max_score credit_performence.indicator_radar = indicator_radar self.performance = credit_performence def credit_analysis_content(): """综合评价分析中信用分析数据""" def business_analysis_content(): """经营分析""" describe = text_model['报告内容'][1]['章节内容'][0]['小节内容'][1]['段落'] return describe def financial_analysis_content(): """财务分析""" if text_model['行业选择'][0] == '制造业': describe = list() describe.append(text_model['报告内容'][3]['章节内容'][0]['小节内容'][1]['段落']) describe.append(text_model['报告内容'][3]['章节内容'][1]['小节内容'][2]['段落']) describe = ''.join(describe) else: describe = text_model['报告内容'][3]['章节内容'][0]['小节内容'][2]['段落'] return describe def risk_analysis_content(): """风险分析""" risk_01 = text_model['报告内容'][4]['章节内容'][0]['小节内容'][0]['段落'] list_01 = risk_01.split('，') risk_02 = text_model['报告内容'][4]['章节内容'][1]['小节内容'][0]['段落'] list_02 = risk_02.split('，') risk_03 = text_model['报告内容'][4]['章节内容'][2]['小节内容'][0]['段落'] list_03 = risk_03.split('，') describe = list() describe.append(list_01[::-1][0]) describe.append(list_02[::-1][0]) describe.append(list_03[::-1][0]) describe = ','.join(describe) res = describe.replace('。', '', 2) return res def evaluation_comments_content(): """评价意见""" describe = list() eva_01 = text_model['报告内容'][5]['章节内容'][0]['小节内容'][0]['段落'] eva_02 = text_model['报告内容'][5]['章节内容'][0]['小节内容'][1]['段落'] eva_03 = text_model['报告内容'][5]['章节内容'][0]['小节内容'][2]['段落'] describe.append(eva_01) describe.append(eva_02) describe.append(eva_03) describe = ''.join(describe) return describe credit_analysis = self.CreditAnalysis() credit_analysis.management_analysis = business_analysis_content() credit_analysis.finance_analysis = financial_analysis_content() credit_analysis.risk_analysis = risk_analysis_content() credit_analysis.evaluation_opinion = evaluation_comments_content() credit_analysis.view_report = '/file/get_cc_rating_report?file_id={}'.format(rating_record['报告fid']) self.credit_analysis = credit_analysis def rating_history(): """历史级别""" df_records['变化趋势'] = (df_records['信用评分'] - df_records['信用评分'].shift(-1)).apply( lambda x: '下降' if x < 0 else ('上升' if x > 0 else ('维持' if x == 0 else '-'))) history_list = list(json.loads(df_records[['信用等级', "信用评分", '评价时间', "评价方式", "变化趋势"]].T.to_json()).values()) self.history_level = list() for item in history_list: history_level = self.HistoryLevel() history_level.credit_rating = item['信用等级'] history_level.credit_score = item['信用评分'] history_level.evaluation_time = item['评价时间'] history_level.evaluation_method = item['评价方式'] history_level.change_trend = item['变化趋势'] self.history_level.append(history_level.dict_to_save()) def rating_index_analysis(): """指标表格""" index_table = self.IndexTable() def financial(): """财务指标""" financial_dict = dict() financial_dict['盈利能力'] = rating_result['财务评分']['盈利能力']['合计'] financial_dict['资产质量'] = rating_result['财务评分']['资产质量']['合计'] financial_dict['债务风险'] = rating_result['财务评分']['债务风险']['合计'] financial_dict['经营增长'] = rating_result['财务评分']['经营增长']['合计'] df_financial = pd.DataFrame( {'实际值': financial_dict, '最大值': {"盈利能力": 16, "资产质量": 18, "债务风险": 18, "经营增长": 18}}) financial_dict = json.loads((df_financial['实际值'] / df_financial['最大值']).apply( lambda x: '优' if x >= 1 else ( '良' if x >= 0.75 else ('中' if x >= 0.5 else ('低' if x >= 0.25 else '差')))).to_json()) financial_index = index_table.FinancialIndicators() financial_index.profitability = financial_dict['盈利能力'] financial_index.asset_quality = financial_dict['资产质量'] financial_index.debt_risk = financial_dict['债务风险'] financial_index.operating_growth = financial_dict['经营增长'] index_table.financial_indicators = financial_index def risks(): """风险指标""" risk_dict = dict() risk_dict['合规风险'] = 43 - rating_result['风险评分']['合规风险'] risk_dict['经营风险'] = 10 - rating_result['风险评分']['经营风险']['合计'] risk_dict['关联风险'] = 10 - rating_result['风险评分']['关联风险']['合计'] df_risks = pd.DataFrame({'实际值': risk_dict, '最大值': {"合规风险": 43, "经营风险": 10, "关联风险": 10}}) risk_dict = json.loads((df_risks['实际值'] / df_risks['最大值']).apply(lambda x: '优' if x >= 1 else ( '良' if x >= 0.75 else ('中' if x >= 0.5 else ('低' if x >= 0.25 else '差')))).to_json()) risk_index = index_table.RiskIndicators() risk_index.compliance_risk = risk_dict['合规风险'] risk_index.operational_risk = risk_dict['经营风险'] risk_index.associated_risk = risk_dict['关联风险'] index_table.risk_indicators = risk_index def operate(): """经营指标""" df_operate = pd.DataFrame( {'实际值': sub_dict(rating_result['经营评分'], ['环境', '社会责任', '公司治理']), '最大值': {'环境': 10, '社会责任': 10, '公司治理': 10}}) operate_dict = json.loads((df_operate['实际值'] / df_operate['最大值']).apply(lambda x: '优' if x >= 1 else ( '良' if x >= 0.75 else ('中' if x >= 0.5 else ('低' if x >= 0.25 else '差')))).to_json()) operate_index = index_table.OperatingIndicators() operate_index.environmental = operate_dict['环境'] operate_index.social_responsibility = operate_dict['社会责任'] operate_index.corporate_governance = operate_dict['公司治理'] index_table.operating_indicators = operate_index financial() risks() operate() self.index_table = index_table def _main_(): result_head() make_radar_chart() credit_analysis_content() rating_history() rating_index_analysis() result = self.dict_to_save() self.db.upsert_single_data( '企业数据', '综合信用分析', {'评价ID': self.rid}, result ) _main_() class FinancialAnalysisImpl(FinancialElementsAnalysis): """财务要素分析类""" db = MongoHelper("tfse_v0.21") def financial_analysis_etl(self): rating_result = self.db.find_single_data( '综信评价数据', '评价结果', {"评价ID": self.rid}, ["企业ID", "财务评分"] ) rating_input = self.db.find_single_data( '综信评价数据', '填报数据', {"评价ID": self.rid}, ["行业选择", "财务填报"] ) industry = rating_input['行业选择'] search = ["年报期", '净资产收益率', '总资产报酬率', '总资产周转率', '应收账款周转率', '存货周转率', '资产负债率', '已获利息倍数', '速动比率', '营业增长率', '总资产增长率', '技术投入比率'] periods = list( pd.DataFrame(rating_input['财务填报']['资产负债表']).sort_values('报告期', ascending=False)['报告期'][0:2].values) df_recent_2year = pd.DataFrame(self.db.find_all_data('综信评价数据', '财指结果', {'企业ID': rating_result['企业ID']}, search)).sort_values('年报期', ascending=False)[0:2] df_this = df_recent_2year[0:1] df_last = df_recent_2year[1:2] data_last = list(json.loads(df_last[['净资产收益率', '存货周转率', '已获利息倍数', '应收账款周转率', '总资产周转率', '总资产增长率', '总资产报酬率', '技术投入比率', '营业增长率', '资产负债率', '速动比率']].T.to_json()).values()) url = "http://api.fecribd.com/tfse_rating/rating/financial_score" headers = {'token': "X0gSlC!YE8jmr2jJr&ilcjS83j!tsoh5", "content-type": "application/json"} data = {"财务指标": data_last, "所属行业": industry} score_last_origin = json.loads(requests.post(url=url, headers=headers, data=json.dumps(data)).text)[ 'result'] score_avg = { "盈利能力": 8, "资产质量": 9, "债务风险": 9, "经营增长": 9 } score_actl = { "盈利能力": rating_result['财务评分']['盈利能力']['合计'], "资产质量": rating_result['财务评分']['资产质量']['合计'], "债务风险": rating_result['财务评分']['债务风险']['合计'], "经营增长": rating_result['财务评分']['经营增长']['合计'] } score_last = { "盈利能力": score_last_origin['盈利能力']['合计'], "资产质量": score_last_origin['资产质量']['合计'], "债务风险": score_last_origin['债务风险']['合计'], "经营增长": score_last_origin['经营增长']['合计'] } df = pd.DataFrame({'今年值': score_actl, '去年值': score_last, "平均值": score_avg}) def result_head(): """基本信息""" self.cid = rating_result['企业ID'] self.report_date = periods[0] self.update_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) def index_detail(): """指标详情""" index_value = json.loads(df_this[ ['净资产收益率', '总资产报酬率', '总资产周转率', '应收账款周转率', '存货周转率', '资产负债率', '已获利息倍数', '速动比率', '营业增长率', '总资产增长率', '技术投入比率']].apply( lambda x: '{}%'.format(x.values[0]) if '率' in x.name or '倍数' in x.name else x.values[0]).T.to_json()) rate = dict() rate['净资产收益率'] = rating_result['财务评分']['盈利能力']['净资产收益率'] / 8 rate['总资产报酬率'] = rating_result['财务评分']['盈利能力']['总资产报酬率'] / 8 rate['总资产周转率'] = rating_result['财务评分']['资产质量']['总资产周转率'] / 6 rate['存货周转率'] = rating_result['财务评分']['资产质量']['存货周转率'] / 6 rate['应收账款周转率'] = rating_result['财务评分']['资产质量']['应收账款周转率'] / 6 rate['资产负债率'] = rating_result['财务评分']['债务风险']['资产负债率'] / 8 rate['已获利息倍数'] = rating_result['财务评分']['债务风险']['已获利息倍数'] / 5 rate['速动比率'] = rating_result['财务评分']['债务风险']['速动比率'] / 5 rate['营业增长率'] = rating_result['财务评分']['经营增长']['营业增长率'] / 8 rate['总资产增长率'] = rating_result['财务评分']['经营增长']['总资产增长率'] / 5 rate['技术投入比率'] = rating_result['财务评分']['经营增长']['技术投入比率'] / 5 df_rate = pd.DataFrame([rate]) index_level = json.loads(df_rate.apply(lambda x: '优' if x.values[0] >= 1 else ( '良' if x.values[0] >= 0.75 else ( '中' if x.values[0] >= 0.5 else ('低' if x.values[0] >= 0.25 else '差')))).to_json()) df_index = pd.DataFrame({'值': index_value, "级别": index_level}) df_list = df_iterrows(df_index) self.index_details = list() for item in df_list: index_details = self.IndexDetails() index_details.index = item[0] index_details.value = item[1] index_details.level = item[2] self.index_details.append(index_details.dict_to_save()) def compare_to_year(): """财务得分年度比较""" self.compare_to_years = list() def score_actl_script(): """今年财务得分""" compare = self.CompareToYears() score_index = FinancialIndexs() score_index.profitability = rating_result['财务评分']['盈利能力']['合计'] score_index.asset_quality = rating_result['财务评分']['资产质量']['合计'] score_index.debt_risk = rating_result['财务评分']['债务风险']['合计'] score_index.business_growth = rating_result['财务评分']['经营增长']['合计'] compare.year = df_this['年报期'][0] compare.index = score_index self.compare_to_years.append(compare.dict_to_save()) def score_last_script(): """去年财务得分""" compare = self.CompareToYears() score_index = FinancialIndexs() score_index.profitability = score_last_origin['盈利能力']['合计'] score_index.asset_quality = score_last_origin['资产质量']['合计'] score_index.debt_risk = score_last_origin['债务风险']['合计'] score_index.business_growth = score_last_origin['经营增长']['合计'] compare.year = df_last['年报期'][1] compare.index = score_index self.compare_to_years.append(compare.dict_to_save()) score_actl_script() score_last_script() def compare_to_industry(): """财务得分同行比较""" compare_industry = self.CompareToIndustry() def company_level(): """公司水平""" company = FinancialIndexs() company.profitability = rating_result['财务评分']['盈利能力']['合计'] company.asset_quality = rating_result['财务评分']['资产质量']['合计'] company.debt_risk = rating_result['财务评分']['债务风险']['合计'] company.business_growth = rating_result['财务评分']['经营增长']['合计'] compare_industry.company_score = company def average_level(): """平均水平""" average = FinancialIndexs() average.profitability = 8 average.asset_quality = 9 average.debt_risk = 9 average.business_growth = 9 compare_industry.average_score = average company_level() average_level() self.compare_to_industry = compare_industry def delta_to_year(): """财务得分较去年变化""" delat_data = json.loads( (df['今年值'] / df['去年值'] - 1).apply(lambda x: round(x * 100, 2)).T.to_json()) delat = FinancialIndexs() delat.profitability = delat_data['盈利能力'] delat.asset_quality = delat_data['资产质量'] delat.debt_risk = delat_data['债务风险'] delat.business_growth = delat_data['经营增长'] self.delta_to_years = delat def delta_to_industry(): """财务扽分交同行差异""" delat_data = json.loads( (df['今年值'] / df['平均值'] - 1).apply(lambda x: round(x * 100, 2)).T.to_json()) delat = FinancialIndexs() delat.profitability = delat_data['盈利能力'] delat.asset_quality = delat_data['资产质量'] delat.debt_risk = delat_data['债务风险'] delat.business_growth = delat_data['经营增长'] self.delta_to_industry = delat def _main_(): result_head() index_detail() compare_to_year() compare_to_industry() delta_to_year() delta_to_industry() result = self.dict_to_save() self.db.upsert_single_data( '企业数据', '财务要素分析', {"评价ID": self.rid}, result ) _main_() class BusinessAnalysisImpl(BusinessElementsAnalysis): """经营要素分析类""" db = MongoHelper("tfse_v0.21") def risk_analysis_etl(self): rating_result = self.db.find_single_data( '综信评价数据', '评价结果', {"评价ID": self.rid}, ["企业ID", "企业名称", "风险评分"] ) risk_data = self.db.find_single_data( '综信评价数据', '风险数据', {"评价ID": self.rid}, ["合规风险", "经营风险"] ) def head_info(): self.cid = rating_result['企业ID'] self.update_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) def risk_relative_score(): """风险分数""" risk_score = rating_result['风险评分']['合计'] self.risk_score = 100 if risk_score / 43 >= 1 else round(risk_score / 43 * 100, 2) def risk_level(): """风险级别""" risk_score = rating_result['风险评分']['合计'] if risk_score >= 43: level = '高' elif risk_score >= 33: level = '较高' elif risk_score >= 23: level = '中等' elif risk_score >= 10: level = '警示' else: level = '低' self.risk_level = level def risk_num(): """风险数量""" self.risk_quantity = sum(risk_data['合规风险'].values()) + sum(risk_data['经营风险'].values()) def is_break_trust(): """列入失信人名单""" self.untrustworthy_list = '是' if risk_data['合规风险']['失信人'] >= 1 else '否' def eligibility_risk(): """合规风险""" def degree_of_impact(param): if param.name in ['失信人', '严重违法']: impact = '严重' elif param.name in ['经营异常', '欠税公告', '税收违法']: impact = '异常' elif param.name in ['立案信息', '行政处罚', '环保处罚']: impact = '中等' else: impact = '' return impact df = pd.DataFrame({'合规风险': risk_data['合规风险']}) df = df.drop(df[df['合规风险'] == 0].index) df['影响程度'] = df.T.apply(lambda x: degree_of_impact(x)) df_list = df_iterrows(df) self.compliance_risk = list() if df_list: for item in df_list: cr = self.ComplianceRisk() cr.genre = item[0] cr.quantity = item[1] cr.affect = item[2] self.compliance_risk.append(cr.dict_to_save()) def operating_risk(): """经营风险""" def degree_of_impact(param): if param.name in ['开庭公告', '法院公告', '诉讼', '送达公告'] and param.values[0] >= 50: impact = '异常' elif param.name in ['开庭公告', '法院公告', '诉讼', '送达公告'] and param.values[0] >= 25: impact = '中等' elif param.name in ['开庭公告', '法院公告', '诉讼', '送达公告'] and param.values[0] >= 10: impact = '警示' elif param.name in ['被执行人'] and param.values[0] >= 30: impact = '异常' elif param.name in ['被执行人'] and param.values[0] >= 15: impact = '中等' elif param.name in ['被执行人'] and param.values[0] >= 6: impact = '警示' elif param.name in ['股权出质'] and param.values[0] >= 10: impact = '异常' elif param.name in ['股权出质'] and param.values[0] >= 5: impact = '中等' elif param.name in ['股权出质'] and param.values[0] >= 3: impact = '警示' else: impact = '轻微' return impact df = pd.DataFrame({'经营风险': risk_data['经营风险']}) df = df.drop(df[df['经营风险'] == 0].index) df['影响程度'] = df.T.apply(lambda x: degree_of_impact(x)) df_list = df_iterrows(df) self.operational_risk = list() if df_list: for item in df_list: or_ = self.OperationalRisk() or_.genre = item[0] or_.quantity = item[1] or_.affect = item[2] self.compliance_risk.append(or_.dict_to_save()) def associate_risk(): """周边风险""" data = associate_risk_detail(rating_result['企业名称']) self.peripheral_risk = list() if data: df_list = df_iterrows(pd.DataFrame(data).sort_values('total', ascending=False).set_index('title')) for item in df_list: pr = self.PeripheralRisk() pr.genre = item[0] pr.quantity = item[1] pr.detail = item[2] self.peripheral_risk.append(pr.dict_to_save()) def change_log(): """变更记录""" data = change_log_detail(rating_result['企业名称']) def degree_of_impact(param): if param.name in ['法定代表人变更', '主要人员变更'] and param.values[0] >= 20: impact = '异常' elif param.name in ['法定代表人变更', '主要人员变更'] and param.values[0] >= 10: impact = '中等' elif param.name in ['法定代表人变更', '主要人员变更'] and param.values[0] >= 4: impact = '警示' else: impact = '轻微' return impact df = pd.DataFrame(data).sort_values('total', ascending=False).set_index('title') df['影响程度'] = df.T.apply(lambda x: degree_of_impact(x)) df_list = df_iterrows(df[['total', '影响程度']]) self.change_log = list() for item in df_list: cl = self.ChangeLog() cl.genre = item[0] cl.quantity = item[1] cl.affect = item[2] self.change_log.append(cl.dict_to_save()) def _main_(): head_info() risk_relative_score() risk_level() risk_num() is_break_trust() eligibility_risk() operating_risk() associate_risk() change_log() result = self.dict_to_save() self.db.upsert_single_data( '企业数据', '经营风险分析', {'评价ID': self.rid}, result ) _main_() class EsgAnalysisImpl(EsgRatingAnalysis): """ESG评价分析类""" db = MongoHelper("tfse_v0.21") def egs_analysis_etl(self): rating_record = self.db.find_single_data( 'ESG评价数据', '评价结果', {"评价ID": self.rid}, ["企业ID", "评价等级", "ESG得分", "环境得分", "社会得分", "公司治理得分"] ) rating_input = self.db.find_single_data( 'ESG评价数据', '填报数据', {"评价ID": self.rid}, ["所属行业"] ) self.cid = rating_record['企业ID'] self.update_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()) def esg_rating(): """esg评级""" rating = self.EsgRating() rating.rating = rating_record['评价等级'] rating.score = rating_record['ESG得分'] rating.industry = rating_input['所属行业'][0] self.esg_rating = rating def dimension_rating(): """维度得分情况""" self.dimension_score = list() def e(): """环境""" dimension = self.DimensionScore() dimension.dimension = '环境' dimension.weights = '35%' dimension.score = rating_record['环境得分']['合计'] dimension.score_change = '-%' dimension.industry_average = '-' self.dimension_score.append(dimension.dict_to_save()) def s(): """社会""" dimension = self.DimensionScore() dimension.dimension = '社会' dimension.weights = '35%' dimension.score = rating_record['社会得分']['合计'] dimension.score_change = '-%' dimension.industry_average = '-' self.dimension_score.append(dimension.dict_to_save()) def g(): """治理""" dimension = self.DimensionScore() dimension.dimension = '治理' dimension.weights = '30%' dimension.score = rating_record['公司治理得分']['合计'] dimension.score_change = '-%' dimension.industry_average = '-' self.dimension_score.append(dimension.dict_to_save()) def _main_(): e() s() g() _main_() def __main__(): esg_rating() dimension_rating() result = self.dict_to_save() self.db.upsert_single_data( "企业数据", "ESG评价分析", {"评价ID": self.rid}, result ) __main__()